自搭的基于Frida一体化采集框架

提示：本文仅分享框架设计思路和大体的使用，全是博主自己个人的设计思路，转载请注明出处

文章目录

• 前言
• 一、cpfrida的功能
• • 1.日志收集
  • 2.设备 *** 作
  • 3.异常告警
  • 4.模板处理
  • 5.进程维护
  • 6.自动运维
  • 7.快速生成项目
  • 8.频率可调控
  • 9.服务端快速部署
  • 10.hook脚本的雏形（自己编写hook逻辑）
  • 11.手机端快速部署（自己实现）
• 二、宏观看架构
• 三、微观看架构
• • 1.服务端环境
  • 2.手机端环境
  • 3.辅助环境
• 四、代码详解
• • 1.项目目录
  • 2.javascript核心代码
  • 3.python的核心代码
• 五、使用步骤
• • 1.安装python依赖
  • 2.配置项目文件build.ini
  • 3.deploy.py的用法
  • 4.项目启动文件说明
  • 5.本地调试
  • 6.远程部署（Jenkins）
  • 7.路径的问题
  • 8.源码
• 总结

---

前言

随着自己做逆向采集已经快一年了，逆向的方向还是半桶水，但是优化采集流程，搭建自己的采集体系方便工作和运维却是在这一年中有较大的提升。本框架就提供一个学习思路和大体的使用，所有的一些 *** 作都是博主个人经验的总结，用代码的形式展现出来而已，细节方面还得靠自己去看代码跟着思路走。下文中的cpfirda就是本次的框架，是基于frida进行的二次开发，好，下面进入正文，害怕起来了。。。。。。

---

一、cpfrida的功能

简单用一句话来说：搭好架构，完善hook脚本，配置好项目配置，手机直接接入，就可以采集数据

1.日志收集

提供了简易的日志收集，用来记录每次hook的数据、数据回传、对设备的 *** 作以及一些异常信息的收集等等，用来分析一个完整的hook流程或者分析hook过程中的一些异常，方便后续优化流程等 *** 作

2.设备 *** 作

提供了简易的运维 *** 作，因为在hook的时候设备或系统的原因，会导致app经常卡死，这时候就有重启手机或者app等简单的 *** 作来保持hook流程的正常进行

3.异常告警

提供了一些消息通知，正常的一个项目，肯定要有自己的告警体系，这边我采用了飞书群通知的形式告知运维人员设备风控等情况，至于要其他形式，需要自己找api接入

4.模板处理

提供了模板快速生成代码的功能，为什么要这个呢？有做过frida采集的同学一定深有体会，frida是用脚本的形式去打桩的，每次脚本可能都不大一样，但是总体又可能相似，就差一些参数或者频率等等，这时候模板就会帮助我们去节省大量的开发时间了

5.进程维护

提供了进程维护功能，frida是脚本形式就奠定了它比较难以有效的集中管理，通常是以脚本的形式运行业务，那一些奇奇怪怪的问题就会导致脚本停止，这时候就需要在脚本运行的时候记录一下当前脚本的进程信息，方便后续运维

6.自动运维

提供了简单的自动运维功能，当脚本停止，数据长时间没有回调，我们是需要一个数据回调的统计和告警平台的，当数据长时间未回调，这个平台上的对应信息就会出现一个告警（这个流程也比较简单，就是监控数据回调的时长，如果超过某个值没有数据回传，那么就认为设备异常，出告警就行了），然后框架会定时去访问这个平台，捕获到告警信息，对对应的设备进行重启等 *** 作，如果框架有在运维对应的进程，就会杀死对应的进程，重新在创建一个新的进程（为什么要杀死？因为如果那个进程还在的话没有什么问题，只是在等待hook的参数，那么当没有杀死进程的情况下，一旦拿到了hook参数，那么此时就有两条进程在同时跑一个设备一个app的打桩点，目前以我的经验这样是行不通的）

7.快速生成项目

基于第四个模板处理下，框架提供了根据配置文件快速生成项目的功能，说白了就是项目开始自有配置文件，框架会根据配置项生成不同的工作空间（这个后续在讲什么是工作空间）来完善整个项目。当然还提供了收纳项目了，会根据框架中你可能手动改过的配置进行更新已有的配置，同时删除了对应的工作空间，这样项目的可移植性非常之高了

8.频率可调控

做爬虫的或者逆向采集的，都知道频率是个很玄学的东西，我们会经常去修改频率来满足我们的业务需求或者来规避风控等等，该框架集成了一些采集的频率，我们可以通过认为的方式去调控频率，但不是实时的，需要停止该业务线在启动才生效（如果实时的我觉得IO应该会比较高。。。。。。）

9.服务端快速部署

提供了一快速部署项目的脚本，如果用Jenkins的话，拉玩代码执行该脚本就行，如果手动的也是一样，十分方便

10.hook脚本的雏形（自己编写hook逻辑）

集成了一些hook过程常用的工具，需要手动拉起对应的dex包到手机的/data/loacl/tmp目录，只有这个目录用户才有 *** 作的权限

11.手机端快速部署（自己实现）

这个我就给个思路吧，这个地方比较复杂，我们开发一个APP，这个APP有什么功能呢？
a、可以发送广播，这个广播可以启动内网穿透对应的程序和frida服务端的程序，而且要实时运行，这样就可以完全打通服务端和手机端的正常通信了
b、这个APP可以释放对应项目所需的dex包和frida-server包以及你们自己的一些脚本啊等的
c、这个APP可以拉起内网穿透程序配置，这个配置是把frida端口和ADB端口映射到服务端的，这样能方便我们工作，手机中去编辑配置，想想都蛋疼。。。。。。

二、宏观看架构

上面介绍完了功能，现在我们来看看项目的整体流程，直接上流程图：

这张图片是最早之前自己画的流程图，其中还包括了开发部分，很简单明了，相信一看就到明白了。传统的Frida在本地调试的时候需要用usb连接，我框架中直接采用远程连接，然后利用frp内网穿透工具把frida的端口和ADB的端口直接都映射到Linux服务器上，这样就可以直接在服务器上的某个端口访问手机端的frida端口和ADB端口实现远程 *** 作了，像之前如何将手机变成一个（Linux）服务器提到的在手机上安装Linux服务，非常之不稳定和繁琐，这样做的稳定性和简便我自己是深有体会的

三、微观看架构 1.服务端环境

先上图：
a、python3.8：为了兼容frida15.1.10版本，这里我选用Anaconda3
b、frida-client：对于手机来说，此时linux就是客户端，frida对应的插件可以网上找一大堆，这里就不赘述了，我选用的是葫芦娃大佬编译的frida
c、adb工具：远程 *** 作手机和app，网上一堆
d、frp-server：映射手机上的frida和adb端口的，网上都有，其他工具只要能实现内网穿透都行
e、linux自带的定时任务crontab：用来定时启动监控脚本，在项目启动文件中已经写好了

2.手机端环境

先上图：

a、fastjson.dex：阿里开发的对象和JSON的处理包
b、FRP_INSTALL_YouWant：自己开发的APP用来发送广播，维护frida和frp的app
c、helper：自己开发的JONS处理工具，删除JSON不需要的字段
d、Magisk：获取手机的root权限，在刷机的时候刷入，可以去玩主页学习怎么刷机点击学习
e、RootExplorer：文件管理工具，系统自带的不好用

3.辅助环境

这里的辅助指的是一件部署功能的，即jenkins，详细的部署流程和简单的使用可以去我主页看对应的博文点击学习

你也许会觉得上面的插件和APP怎么这么多啊，这个就看你自己规划了，可以根据自己的认知简易部署流程，或者可以舍弃一些自己觉得没用的东西，毕竟有些东西是辅助作用，不是项目的核心需要哈

四、代码详解 1.项目目录

项目目录这个东西啊，真的有些头疼，开发的小伙伴也行会感同身受，本项目的目录大部分以及配置好了，无需太多的改动，上项目目录图：

cpfrida：项目名称

v1.0：版本，和v2.0采集逻辑上不同，v1.0是采用权限分发的形式，怎么分发的呢？这么说吧，你自顾自的采集会导致数据失衡，满足不了业务，所有规则就是满足业务需求，举个例子：比如有几个接口detail、product、live、topic，业务需要的是detail一个，product三个，live五个，topic两个，这时候假设你要部署11个设备上去满足业务，但是在采集的过程中，你会发现某一天product数据量过多，在规定时间内完成不了，但是呢，live数据量过少很大一部分时间会处于空闲时间，这样就相当于浪费了资源，于是就有了权限分发，每台设备先按之前的方式部署好，detail一个，product三个，live五个，topic两个，当某台设备空闲的时候就会执行下级权限，现在live的设备权限顺序为：live–>product–>detail–>topic，当live空闲的时候，它就执行product，product也空闲的时候依次类推，每次完成一次完整的hook之后，权限又是重头开始，这样做就可以达到资源全部合理利用啦，对了，牢记frida无法同时在一个点打桩多次，按我hook的app来看的话是这样，上面的权限分发就有效的避开了这个问题

v2.0：版本，和v1.0采集逻辑上不同，v2.0是采用进程池的概念，它比一来的简单暴力，但是业务线过多的话Linux服务器可能就需要多个了，比较耗费一些资源。它没有那些采集规则，业务不够就加一套，还是接口detail、product、live、topic，一样的部署方式，只是此时11个设备我每个设备都跑四条业务，这时候就有44个业务了，是不是体量多太多了，想想都有点小刺激

common：一些通用模块

• build_model.py：项目配置文件模块
• handle_template.py：模板生成代码核心
• ident_model.py：工作空间标识
• inform_mode.py：消息通知模块，对接群机器人等api
• logging_mode.py：日志模块
• phone_model.py：手机 *** 作模块
• process_model.py：进程管理模块

deploy：快速生成项目相关，已经封装在项目部署文件中

• template：代码模板

  • config_template：每个工作空间的配置项
  • startwork_template：工作空间启动项，sh
  • workspace_template：工作空间代码

• build.ini：项目配置项，所有的工作空间都是根据这个配置项生成的

• build_.ini：配置项的各项说明

• deploy.py：快速生成代码的核心，已集成到项目启动脚本中

hookscript：hook脚本js代码

• vx_hooker_8015.js：hook代码的逻辑，已经封装好了一些常用的工具
• vx_hooker_8018.js：hook代码的逻辑，已经封装好了一些常用的工具

startproject：项目部署启动相关

• kill_process.py：杀死当前系统下的说有python进程，所有建议服务器单一点就跑这个采集框架，v2.0是这样的，v1.0比较智能，是根据进程管理来进行kill的，也可能错杀，当没有v2.0这么暴力
• reboot_phone.py：重启所有的设备，多进程模式重启
• start_project.py：解压项目启动脚本，sh，因为直接sh通过jenkins部署会存在编码问题

workspace：工作空间

工作空间下自由的模块：

• monitor：监控模块
  • monitor.py：监控基类，封装好了一些常用方法和 *** 作
  • monitor.py：继承监控基类，是点对点的，专属我自己的一个监控，可以自行编写哈
• base_script.py：frida的核心创建逻辑、对外部分交互和用到的模块载入
• business.py：hook的业务逻辑和send数据处理
• extend.py：项目的扩展功能，比如收集cookie，或者网页端采集等等
• params.py：项目用到的固有参数，和一些任务常调控的参数，比如频率
• device1：工作空间（设备名称），这个是根据build.ini生成出来的，一个完整的工作空间是有：
  • 配置文件 config.py，对应的配置项都有对应的说明
  • 工作者 device1.py，单个进程或进程池的入口
  • 工作空间启动项 device1.sh，值适用linux系统，window系统可以直接手动输入指令执行或者在IDE中执行
  • 工作空间身份 uuid

requirements.txt：项目依赖

2.javascript核心代码

不会反编译，一切都是徒劳的，下面就给出了js的辅助工具，至于hook的核心代码和接口，这边不便展示，懂得都懂（手动狗头）：

/**
 * @Description: Wechat 8015 hook script
 * @author XQE
 * @date 2022/04/19
*/


var tools = {
    classexists: function (className) {
        var _exists = false
        try {
            Java.use(className)
            _exists = true
        } catch (err) {
        }
        return _exists
    },

    checkloaddex: function (className, dexfile) {
        if (!this.classexists(className)) {
            Java.openClassFile(dexfile).load()
        }
    },

    tojsonstring: function (obj) {
        try {
            this.checkloaddex("com.alibaba.fastjson.JSON", "/data/local/tmp/fastjson.dex")
            var _clz = Java.use("com.alibaba.fastjson.JSON")
            var _toJSONStringMehtod = _clz.toJSONString.overload("java.lang.Object")
            return _toJSONStringMehtod.call(_clz, obj)
        } catch (err) {
            console.log(err)
        }
        return "{}"
    },

    fromjsonstring: function (jsonStr, classObj = null) {
        try {
            this.checkloaddex("com.alibaba.fastjson.JSON", "/data/local/tmp/fastjson.dex")
            if (classObj == null) {
                var _clz = Java.use("com.alibaba.fastjson.JSON")
                return _clz.parseObject(jsonStr)
            } else {
                var _jsonObject = Java.use("com.alibaba.fastjson.JSONObject")
                var _obj = _jsonObject.parseObject(jsonStr, classObj.class)
                return _obj
            }
        } catch (err) {
            console.log(err)
        }
        return null
    },

    recursionremove: function (jsonObject, removeKey) {
        var _keyArray = jsonObject.keySet().toArray()
        for (var i = 0; i < _keyArray.length; i++) {
            var _key = _keyArray[i]
            var _object = jsonObject.get(_key)
            var _objectType = ""
            try {
                _objectType = _object.getClass().getName()
            } catch (e) {}
            if (_key == removeKey && _objectType == "java.lang.String") {
                jsonObject.remove(_key)
            }
            if (_objectType == "com.alibaba.fastjson.JSONObject") {
                this.recursionremove(jsonObject.getJSONObject(_key), removeKey)
            }
            if (_objectType == "com.alibaba.fastjson.JSONArray") {
                var jsonArray = jsonObject.getJSONArray(_key)
                for (var a = 0; a <= jsonArray.size(); a++) {
                    try {
                        var _childObj = jsonArray.getJSONObject(a)
                        if (_childObj) {
                            this.recursionremove(_childObj, removeKey)
                        }
                    } catch (e) {
                        console.log(e)
                    }
                }
            }
        }
    },

    wxhelper: function () {
        try {
            this.checkloaddex("com.alibaba.fastjson.JSON", "/data/local/tmp/fastjson.dex")
            this.checkloaddex("com.tencent.mm.wechathelperdex.WxHelper", "/data/local/tmp/wxhelper.dex")
            var _clz = Java.use("com.tencent.mm.wechathelperdex.WxHelper")
            return _clz
        } catch (err) {
            console.log(err)
        }
        return "{}"
    },

    hashset: function () {
        var _HashSetClz = Java.use("java.util.HashSet")
        var _hashset = _HashSetClz.$new()
        return _hashset
    },

    map: function () {
        var _hashMap_clz = Java.use("java.util.HashMap")
        var _map_clz = Java.use("java.util.Map")
        var _hashMap = _hashMap_clz.$new()
        var _map = Java.cast(_hashMap, _map_clz)
        return _map
    },

    base64: function (zsBase64) {
        var androidBase64 = Java.use("android.util.Base64")
        var _base64 = androidBase64.decode(zsBase64, 0)
        return _base64
    },
}

var hooker_fun = {
    runComment: function (device, reply, args) {},

    runDetailVideo: function (device, args) {},

    runDetailGoods: function (device, args) {},

    runLiveInfo: function (device, username, args) {},

    runLiveBarrage: function (device, username, args) {},
	
	runLiveContribution: function(device, username, args) {},

    runLiveGoods: function (device, username, args) {},

    runLiveSquare: function (device, cate) {},

    runLiveTab: function () {},

    runLiveBag: function (device, args) {},

    runProductList: function (device, selfappid, args) {},

    runProductInfo: function (device, selfappid, args) {},

    runProductStore: function (device, selfappid, args) {},

    runProductTakecenter: function (device, selfappid, cookie, args) {}

    runProductThird: function (device, selfappid, cookie, args) {},

    runGetcookie: function (device, username) {},

    runTopicTopic: function (device, args) {},

    runTopicActivity: function (device, tab, args) {},

    runVideoGoods: function (device, args) {},
    
    runVideoUrl: function (device, args) {},
}

rpc.exports = {
    hookerVideoComment: function (device, reply, args) {
        Java.perform(function () {
            hooker_fun.runComment(device, reply, args)
        })
    },

    hookerDetailVideo: function (device, args) {
        Java.perform(function () {
            hooker_fun.runDetailVideo(device, args)
        })
    },

    hookerDetailGoods: function (device, args) {
        Java.perform(function () {
            hooker_fun.runDetailGoods(device, args)
        })
    },

    hookerLiveInfo: function (device, username, args) {
        Java.perform(function () {
            hooker_fun.runLiveInfo(device, username, args)
        })
    },

    hookerLiveBarrage: function (device, username, args) {
        Java.perform(function () {
            hooker_fun.runLiveBarrage(device, username, args)
        })
    },

    hookerLiveContribution: function (device, username, args) {
        Java.perform(function () {
            hooker_fun.runLiveContribution(device, username, args)
        })
    },

    hookerLiveGoods: function (device, username, args) {
        Java.perform(function () {
            hooker_fun.runLiveGoods(device, username, args)
        })
    },

    hookerLiveSquare: function (device, cate) {
        Java.perform(function () {
            hooker_fun.runLiveSquare(device, cate)
        })
    },

    hookerLiveTab: function () {
        Java.perform(function () {
            hooker_fun.runLiveTab()
        })
    },

    hookerLiveBag: function (device, args) {
        Java.perform(function () {
            hooker_fun.runLiveBag(device, args)
        })
    },

    hookerProductList: function (device, selfappid, args) {
        Java.perform(function () {
            hooker_fun.runProductList(device, selfappid, args)
        })
    },

    hookerProductInfo: function (device, selfappid, args) {
        Java.perform(function () {
            hooker_fun.runProductInfo(device, selfappid, args)
        })
    },

    hookerProductStore: function (device, selfappid, args) {
        Java.perform(function () {
            hooker_fun.runProductStore(device, selfappid, args)
        })
    },

    hookerProductTakecenter: function (device, selfappid, cookie, args) {
        Java.perform(function () {
            hooker_fun.runProductTakecenter(device, selfappid, cookie, args)
        })
    },

    hookerProductThird: function (device, selfappid, cookie, args) {
        Java.perform(function () {
            hooker_fun.runProductThird(device, selfappid, cookie, args)
        })
    },

    hookerGetcookie: function (device, username) {
        Java.perform(function () {
            hooker_fun.runGetcookie(device, username)
        })
    },

    hookerTopicTopic: function (device, args) {
        Java.perform(function () {
            hooker_fun.runTopicTopic(device, args)
        })
    },

    hookerTopicActivity: function (device, tab, args) {
        Java.perform(function () {
            hooker_fun.runTopicActivity(device, tab, args)
        })
    },

    hookerVideoGoods: function (device, args) {
        Java.perform(function () {
            hooker_fun.runVideoGoods(device, args)
        })
    },

    hookerVersionCheck: function () {
        Java.perform(function () {
            hooker_fun.runVersionCheck()
        })
    },

    hookerVideoUrl: function (device, args) {
        Java.perform(function () {
            hooker_fun.runVideoUrl(device, args)
        })
    },
}

讲解：
从代码中我们可以看到总体有三部分：

• tools：封装了一些常用的工具，其中包括
  • classexists：判断某个引用类是否存在
  • checkloaddex：加载dex包
  • tojsonstring：对象转json
  • fromjsonstring：json转对象
  • recursionremove：js层面删除json某个字段
  • wxhelper：java层面删除json某个字段，比js快
  • hashset：创建hashset容器
  • map：创建map容器
  • base64：字符串转android的base64
• hooker_fun：hook的逻辑实现处，这里仁者见仁智者见智，自行发挥，这部分太核心了，不方便展示
• rpc.exports：rpc远程调用，这个是一种写法，网上还有蛮多写法的，可以自行查阅，要注意的是这里的函数名要和python里的frida名称一致，但是比如hookerVideoComment，python里面要这样的hooker_video_comment

3.python的核心代码

A、business：采集流程控制核心

# !/usr/bin/env python
# coding=utf-8
# @Time    : 2022/04/20
# @Author  : XQE
# @Software: PyCharm


import os
import sys
import time
import json
import random
import config
import requests
sys.path.append("../common")
sys.path.append("../../common")
from extend import Extend
from base_script import BaseScript
from inform_model import Inform


class AdapterBusiness(Extend, BaseScript, Inform):
    """
    对hook业务进行适配，分发到对应的业务链上进行hook
    """
    Adapter = dict()

    def __init__(self, filename, device, url, full_path=""):
        Extend.__init__(self)
        BaseScript.__init__(self, filename, device, url, full_path)
        Inform.__init__(self, config.inform_url)
        for i in dir(self):
            self.Adapter[i] = getattr(self, i)

    def hooker_video_url(self):
        """
        hook视频播放地址
        """
        self.frida_object.exports.hooker_video_url(
            self.device,
            self.get_response
        )

    def hooker_video_goods(self):
        """
        hook视频挂链
        """
        self.frida_object.exports.hooker_video_goods(
            self.device,
            self.get_response
        )

    def hooker_topic_activity(self, tab=1):
        """
        hook视频活动
        """
        self.frida_object.exports.hooker_topic_activity(
            self.device,
            tab,
            self.get_response
        )

    def hooker_topic_topic(self):
        """
        hook视频话题
        """
        self.frida_object.exports.hooker_topic_topic(
            self.device,
            self.get_response
        )

    def hooker_getcookie(self):
        """
        hook微信cookie
        """
        self.frida_object.exports.hooker_getcookie(
            "cookie",
            "v2_060000231003b20faec8c5e08a18c4dcc60cec31b077b2e0eab771d78025b3800ae7b330f30e@finder"
        )

    def hooker_product_third(self):
        """
        hook带货中心商品详情
        """
        self.frida_object.exports.hooker_product_third(
            self.device,
            "wx2cea70df4257bba8",
            self.vx_cookie,
            self.get_response
        )

    def hooker_product_takecenter(self):
        """
        hook带货中心商品列表
        """
        self.frida_object.exports.hooker_product_takecenter(
            self.device,
            "wx2cea70df4257bba8",
            self.vx_cookie,
            self.get_response
        )

    def hooker_product_store(self):
        """
        hook微信小商店信息
        """
        self.frida_object.exports.hooker_product_store(
            self.device,
            "wx34345ae5855f892d",
            self.get_response
        )

    def hooker_product_info(self):
        """
        hook微信小商店商品详情
        """
        self.frida_object.exports.hooker_product_info(
            self.device,
            "wx34345ae5855f892d",
            self.get_response
        )

    def hooker_product_list(self):
        """
        hook微信小商店商品列表
        """
        self.frida_object.exports.hooker_product_list(
            self.device,
            "wx34345ae5855f892d",
            self.get_response
        )

    def hooker_live_tab(self):
        """
        hook直播分类
        """
        self.frida_object.exports.hooker_live_tab(self.device)

    def hooker_live_square(self):
        """
        hook直播广场
        """
        for i in self.get_response.get("data").get("cates"):
            if i.get("cateid") > 10:
                continue
            self.check_frida()
            self.frida_object.exports.hooker_live_square(
                self.device,
                {
                    "VHZ": i.get("cateid"),
                    "VIa": i.get("catename"),
                    "VIb": False,
                    "VIc": False,
                    "VId": False,
                    "VIe": False,
                    "VIf": [],
                    "VIg": 0,
                    "data": "",
                    "includeUnKnownField": False,
                    "object_id": 0
                }
            )
            self.wait()

    def hooker_live_goods(self):
        """
        hook直播带货
        """
        self.frida_object.exports.hooker_live_goods(
            self.device,
            self.username,
            self.get_response
        )

    def hooker_live_contribution(self):
        """
        hook直播间热度贡献榜
        """
        self.frida_object.exports.hooker_live_contribution(
            self.device,
            self.username,
            self.get_response
        )

    def hooker_live_barrage(self):
        """
        hook直播间d幕
        """
        self.frida_object.exports.hooker_live_barrage(
            self.device,
            self.username,
            self.get_response
        )

    def hooker_live_info(self):
        """
        hook直播间信息
        """
        self.frida_object.exports.hooker_live_info(
            self.device,
            self.username,
            self.get_response
        )

    def hooker_detail_goods(self):
        """
        hook主页商品
        """
        self.frida_object.exports.hooker_detail_goods(
            self.device,
            self.get_response
        )

    def hooker_detail_video(self):
        """
        hook主页视频
        """
        self.frida_object.exports.hooker_detail_video(
            self.device,
            self.get_response
        )

    def hooker_video_comment(self):
        """
        hook视频评论
        """
        self.frida_object.exports.hooker_video_comment(
            self.device,
            False,
            self.get_response
        )

    def adapter_business(self):
        """
        hook业务分发
        """
        self.check_frida()
        self.get_params()
        if self.device == "comment":
            self.Adapter["hooker_video_comment"]()
            self.wait()
        elif self.device == "detail":
            self.Adapter["hooker_detail_video"]()
            self.wait()
            self.check_frida()
            self.Adapter["hooker_detail_goods"]()
            self.wait()
        elif self.device == "hourlist":
            if not (self.vx_cookie and os.path.exists(self.cookie_file)):
                self.get_cookie()
                self.wait_cookie()
            content = self.make_hourlist_response(self.get_response)
            self.post_response = content
            self.Handle[self.device](config.post_url.get(self.device).format(self.uuid, self.addr))
        elif self.device == "livebarrage":
            self.Adapter["hooker_live_barrage"]()
            self.wait()
        elif self.device == "liveinfo":
            self.Adapter["hooker_live_info"]()
            self.wait()
        elif self.device == "livecontribution":
            self.Adapter["hooker_live_contribution"]()
            self.wait()
        elif self.device == "livegoods":
            self.Adapter["hooker_live_goods"]()
            self.wait()
        elif self.device == "livesquare":
            self.Adapter["hooker_live_tab"]()
            self.wait()
            self.Adapter["hooker_live_square"]()
        elif self.device == "product":
            if self.get_response.get("data").get("method") == "GetShopCenter":
                self.Adapter["hooker_product_store"]()
            elif self.get_response.get("data").get("method") == "BatchGetProductByBizUin":
                self.Adapter["hooker_product_list"]()
            elif self.get_response.get("data").get("method") == "GetProduct":
                self.Adapter["hooker_product_info"]()
            elif self.get_response.get("data").get("method") == "EcGetItemList":
                if self.vx_cookie:
                    self.Adapter["hooker_product_takecenter"]()
                else:
                    self.get_cookie()
                    self.wait_cookie()
                    self.check_frida()
                    self.Adapter["hooker_product_takecenter"]()
            elif self.get_response.get("data").get("method") == "EcGetItemDetail":
                if self.vx_cookie:
                    self.Adapter["hooker_product_third"]()
                else:
                    self.get_cookie()
                    self.wait_cookie()
                    self.check_frida()
                    self.Adapter["hooker_product_third"]()
            else:
                self.logger.error("product business undefined method %s ." % self.get_response.get("data").get("method"))
            self.wait()
        elif self.device == "topic":
            if int(self.get_response["data"].get("type")) == 1:
                self.Adapter["hooker_topic_topic"]()
            elif int(self.get_response["data"].get("type")) == 2:
                self.Adapter["hooker_topic_activity"](1)
                self.wait()
                self.check_frida()
                self.Adapter["hooker_topic_activity"](2)
            else:
                self.logger.error("topic business undefined type %s ." % self.get_response["data"].get("type"))
            self.wait()
        elif self.device == "videogoods":
            self.Adapter["hooker_video_goods"]()
            self.wait()
        elif self.device == "videourl":
            self.Adapter["hooker_video_url"]()
            self.wait()
        else:
            self.logger.error("can't find this device: %s" % self.device)
        time.sleep(
            random.randint(
                self.sleep_time.get(self.device).get("begin"),
                self.sleep_time.get(self.device).get("end")
            )
        )


class HandleBusiness(AdapterBusiness):
    """
    对send回来的数进行适配，分发到对应的业务链上处理
    """
    Handle = dict()

    def __init__(self, filename, device, url, full_path=""):
        AdapterBusiness.__init__(self, filename, device, url, full_path)
        for i in dir(self):
            self.Handle[i] = getattr(self, i)

    def cookie(self, post_url):
        """
        微信cookie
        :param post_url : 数据回调地址
        """
        self.flag = True
        self.wait_count = 0
        self.remove_monitor = False
        self.get_cookie(self.post_response.get("result").get("VWA"))

    def comment(self, post_url):
        """
        视频评论
        :param post_url : 数据回调地址
        """
        status = requests.post(post_url, json.dumps(self.post_response).encode("utf-8").decode("latin1"))
        self.flag = True
        self.wait_count = 0
        self.remove_monitor = False
        self.logger.info(str(self.post_response.get("device")) + str(status))

    def detail(self, post_url):
        """
        主页信息
        :param post_url : 数据回调地址
        """
        status = requests.post(post_url, json.dumps(self.post_response).encode("utf-8").decode("latin1"))
        self.flag = True
        self.wait_count = 0
        self.remove_monitor = False
        self.logger.info(str(self.post_response.get("device")) + str(status))

    def hourlist(self, post_url):
        """
        小时榜
        :param post_url : 数据回调地址
        """
        for c in self.post_response:
            c["username"] = self.get_response["data"].get("username")
            c["curtime"] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(int(time.time())))
            status = requests.post(post_url, json.dumps(c, ensure_ascii=False).encode("utf-8").decode("latin1"))
            self.logger.info(str(c.get("device")) + str(status))

    def livebarrage(self, post_url):
        """
        直播间d幕
        :param post_url : 数据回调地址
        """
        status = requests.post(post_url, json.dumps(self.post_response).encode("utf-8").decode("latin1"))
        self.flag = True
        self.wait_count = 0
        self.remove_monitor = False
        self.logger.info(str(self.post_response.get("device")) + str(status))

    def liveinfo(self, post_url):
        """
        直播间信息
        :param post_url : 数据回调地址
        """
        status = requests.post(post_url, json.dumps(self.post_response).encode("utf-8").decode("latin1"))
        if self.device == self.post_response.get("device"):
            self.flag = True
        self.wait_count = 0
        self.remove_monitor = False
        self.logger.info(str(self.post_response.get("device")) + str(status))

    def livecontribution(self, post_url):
        """
        直播间热度贡献榜
        :param post_url : 数据回调地址
        """
        status = requests.post(post_url, json.dumps(self.post_response).encode("utf-8").decode("latin1"))
        self.flag = True
        self.wait_count = 0
        self.remove_monitor = False
        self.logger.info(str(self.post_response.get("device")) + str(status))

    def livegoods(self, post_url):
        """
        直播间带货
        :param post_url : 数据回调地址
        """
        status = requests.post(post_url, json.dumps(self.post_response).encode("utf-8").decode("latin1"))
        self.flag = True
        self.wait_count = 0
        self.remove_monitor = False
        self.logger.info(str(self.post_response.get("device")) + str(status))

    def livesquare(self, post_url):
        """
        直播广场
        :param post_url : 数据回调地址
        """
        status = requests.post(post_url, json.dumps(self.post_response).encode("utf-8").decode("latin1"))
        if int(self.post_response.get("type")) == 2:
            if int(self.post_response.get("hasnext")) == 0:
                self.flag = True
            else:
                self.wait_count = 0
        else:
            self.flag = True
            self.wait_count = 0
        self.logger.info(str(self.post_response.get("device")) + str(status))

    def product(self, post_url):
        """
        微信小商店和带货中心
        :param post_url : 数据回调地址
        """
        if self.post_response.get("method") == "EcGetItemList" or self.post_response.get("method") == "EcGetItemDetail":
            if self.post_response.get("flag") in ["-330", "-333", "-334"] and os.path.exists(self.cookie_file):
                os.remove(self.cookie_file)
                self.vx_cookie = None
                self.logger.info("the cookie has expired. obtain it again.")
                self.flag = True
            else:
                status = requests.post(post_url, json.dumps(self.post_response).encode("utf-8").decode("latin1"))
                self.flag = True
                self.logger.info(str(self.post_response.get("device")) + str(status))
        else:
            if self.post_response.get("method") == "BatchGetProductByBizUin" and self.post_response.get("next_key") == "" and int(
                    self.post_response.get("page_num")) > 0:
                self.flag = True
            elif self.post_response.get("method") == "cookie":
                self.get_cookie(self.post_response.get("result").get("VWA"))
                self.flag = True
            else:
                status = requests.post(post_url, json.dumps(self.post_response).encode("utf-8").decode("latin1"))
                if self.post_response.get("method") == "GetProduct" or self.post_response.get("method") == "GetShopCenter":
                    self.flag = True
                self.logger.info(str(self.post_response.get("device")) + str(status))
        self.wait_count = 0
        self.remove_monitor = False

    def topic(self, post_url):
        """
        话题和活动
        :param post_url : 数据回调地址
        """
        status = requests.post(post_url, json.dumps(self.post_response).encode("utf-8").decode("latin1"))
        if int(self.post_response.get("continueFlag")) == 0:
            self.flag = True
        else:
            self.wait_count = 0
        self.remove_monitor = False
        self.logger.info(str(self.post_response.get("device")) + str(status))

    def videogoods(self, post_url):
        """
        视频挂链
        :param post_url : 数据回调地址
        """
        status = requests.post(post_url, json.dumps(self.post_response).encode("utf-8").decode("latin1"))
        self.flag = True
        self.wait_count = 0
        self.remove_monitor = False
        self.logger.info(str(self.post_response.get("device")) + str(status))

    def videourl(self, post_url):
        """
        视频播放地址
        :param post_url : 数据回调地址
        """
        status = requests.post(post_url, json.dumps(self.post_response).encode("utf-8").decode("latin1"))
        self.flag = True
        self.wait_count = 0
        self.remove_monitor = False
        self.logger.info(str(self.post_response.get("device")) + str(status))

    def handle_business(self):
        """
        send数据适配主逻辑
        """
        try:
            post_url = config.post_url.get(self.post_response.get("device")).format(self.uuid, self.addr)
        except Exception as e:
            self.logger.info("currently, it is an internal service")
            post_url = None
        self.Handle[self.post_response.get("device")](post_url)

以上就是我自己项目的hook核心逻辑和数据处理核心逻辑，大体上来讲就是：从队列拿到hook参数，匹配hook点，对hook到的数据进行处理和回调。

• AdapterBusiness：对hook业务进行适配，分发到对应的业务链上进行hook。“adapter_business”该方法是hook的核心，可以看下对应的逻辑：首先self.check_frida()检测frida，如果没有会创建frida对象；然后看书获取hook参数，这里先检测frida再拿参数是为了避免参数流失过多；最后开始匹配hook点，比如“comment”举例：它会匹配“hooker_video_comment”这个hook点，进入方法内部self.frida_object.exports.hooker_video_comment()实现rpc远程调用（好玄学的），进入到javascript中的hook逻辑，最终会把hook到的数据send出去，而python的流程就会被self.wait()这个方法卡住，等待send回来的数据进行分析

• HandleBusiness：对send回来的数进行适配，分发到对应的业务链上处理。“handle_business”该方法时数据处理的核心，他会根据send回来的“device”字段进行匹配回调业务，比如“comment”继续举例：此时send回来的数据已被self.post_response接收，回调完成后要让self.wait()方法放行进行下一次hook，于是就设置self.flag、self.wait_count、self.remove_monitor，使其放行，这三个的具体作用源码中有讲述

B、base_script：frida的核心创建逻辑、对外部分交互和用到的模块载入

# !/usr/bin/env python
# coding=utf-8
# @Time    : 2022/04/20
# @Author  : XQE
# @Software: PyCharm


"""
整个采集脚本的基础类
完整的流程：
1.hook加载类(js加载数据)
2.hook数据类(js获取数据)
"""

import re
import sys
import time
import frida
import config
import requests
sys.path.append("../common")
sys.path.append("../../common")
from logging_model import Logging
from process_model import Process
from ident_model import Ident
from phone_model import Phone


class BaseScript(Logging, Process, Ident, Phone):
    flag = False
    device = ""
    wait_count = 0
    frida_object = None
    get_response = dict()
    post_response = dict()
    remove_monitor = False

    def __init__(self, filename, device, url, full_path=""):
        """
        初始化对象：
        :param filename : 工作空间文件名
        :param full_path : 完整路径
        """
        self.device = device
        self.filename = filename
        self.timeout = config.timeout
        self.frida_host = config.frida_host
        self.app_name_en = config.app_name_en
        self.app_name_ch = config.app_name_ch
        self.hook_script = config.hook_script
        Logging.__init__(self, filename, device, full_path)
        Process.__init__(self, filename, device + re.search(r"\d+", filename).group(), "../monitor/monitor_conf.ini", full_path)
        Ident.__init__(self, filename, full_path)
        Phone.__init__(self, config.adb_host, self.app_name_en, config.app_ui)
        self.addr = self.get_host_ip()
        self.uuid = self.get_workspace_uuid()
        self.get_params_url = url

    def check_frida(self):
        """
        检测frida
        """
        if self.frida_object is None:
            self.frida_object = self.session()

    def session(self):
        """
        frida远程植入微信进程
        :return: 植入载体
        """
        session = None
        while session is None:
            device = None
            try:
                manager = frida.get_device_manager()
                device = manager.add_remote_device(self.frida_host)
            except Exception as e:
                self.logger.error("frida connect fail: " + str(e), exc_info=False)
            if device is not None:
                try:
                    session = device.attach(self.app_name_en)
                    self.logger.info("frida attach appname %s" % self.app_name_en)
                except Exception as e:
                    self.logger.error("frida attach %s fail: " % self.app_name_en + str(e), exc_info=False)
                    try:
                        session = device.attach(self.app_name_ch)
                        self.logger.info("frida attach appname %s" % self.app_name_ch)
                    except Exception as e:
                        self.logger.error("frida attach %s fail: " % self.app_name_ch + str(e), exc_info=False)
                        time.sleep(5)
        try:
            with open(self.hook_script, encoding="utf-8") as f:
                script = session.create_script(f.read(), runtime="v8")
                script.on("message", self.my_message_handler)
                script.load()
                self.logger.info("hook script success.")
                return script
        except Exception as e:
            self.logger.error("frida create script fail: " + str(e), exc_info=False)

    def wait(self):
        """
        限时等待js响应
        """
        self.wait_count = 0
        self.flag = False
        while True:
            time.sleep(0.1)
            self.wait_count += 1
            if self.flag or self.wait_count > self.timeout:
                if self.wait_count > self.timeout:
                    self.logger.info("hook time out ...")
                    # 当存在风控风险时重启设备
                    if self.remove_monitor:
                        self.logger.info("reboot phone ...")
                        self.reboot()
                    if self.frida_object:
                        self.frida_object.unload()
                    break
                self.flag = False
                break

    def get_params(self):
        """
        获取队列消息
        """
        while True:
            try:
                self.get_response = requests.get(self.get_params_url.format(self.uuid, self.addr))
                self.get_response.encoding = "utf-8"
                self.get_response = self.get_response.json()
                self.logger.info(self.get_response)
                if self.get_response.get("code") == 0:
                    self.logger.info("the current business: %s" % self.device)
                    break
                time.sleep(1)
            except Exception as e:
                self.logger.error("params queue closed: " + str(e), exc_info=False)

• frida的创建：”session“方法，如果没有创建成功，会一直卡在这里，不会消耗队列的数据，这里创建的时远程连接
• 对外交互：“get_params”方法，获取队列中的hook参数，v1.0是设置权限的，v2.0是进程池的，代码有点差距
• init：初始化方法，加载的所有相关的模块都在这实现了，少部分会在business.py中加载
• wait：等待hook回传数据，比较灵活，可以看为一个进程锁

五、使用步骤

以下的前提是：搭好了架构和环境

1.安装python依赖

根据requirements.txt文件中的版本安装好第三方库，指令在该文件末尾，安装时候记得删除最后两行的指令（建议使用外网安装，不然就自己手动网上去找对应的frida==15.1.10的第三方库）

2.配置项目文件build.ini

# 项目全局配置
[project]
# 项目目录
path=D:\\工作目录\\cpFrida\\v1.0

# python环境
[python_v]
# python版本
python=3.8

# 监控配置
[monitor]
# 重启时间差
reboot=180

# 工作空间配置
[comment1]
# py等待js最大send时长
timeout=150
# frida IP端口
frida_host=114.114.114.114:10406
# adb IP端口
adb_host=114.114.114.114:10706
# app英文包名
app_name_en=com.tencent.mm
# app中文包名
app_name_ch=微信
# app首页ui
app_ui=com.tencent.mm/com.tencent.mm.ui.LauncherUI
# 工作空间名称
workspace=comment1
# hook脚本
hook_script=vx_hooker_8015.js
# 队列池
get_url={"comment": "http://114.114.114.114:8182/comment/8015/task?uuid={}&host={}", "detail": "http://114.114.114.114:8181/detail/8015/task?uuid={}&host={}", "livebarrage": "http://114.114.114.114:8187/livedanmu/8015/task?uuid={}&ip={}", "liveinfo": "http://114.114.114.114:8187/live/8015/task?uuid={}&host={}", "livecontribution": "http://114.114.114.114:8187/liveonlineuser/8015/task?uuid={}&host={}", "livegoods": "http://114.114.114.114:8187/liveproduct/8015/task?uuid={}&host={}", "livesquare": "http://114.114.114.114:8186/livesquare/8015/task?uuid={}&host={}", "product": "http://114.114.114.114:8180/product/8015/task?uuid={}&ip={}", "topic": "http://114.114.114.114:8181/topic/8015/task?uuid={}&host={}", "videogoods": "http://114.114.114.114:8185/videoproduct/8015/task?uuid={}&host={}", "hourlist": "http://114.114.114.114:8186/liverank/8015/task?uuid={}&host={}", "videourl": "http://114.114.114.114:8184/videourl/7012/task?uuid={}&host={}"}
# 回调池
post_url={"comment": "http://114.114.114.114:8182/comment/8015/callback?uuid={}&host={}", "detail": "http://114.114.114.114:8181/detail/8015/callback?uuid={}&host={}", "livebarrage": "http://114.114.114.114:8187/livedanmu/8015/callback?uuid={}&ip={}", "liveinfo": "http://114.114.114.114:8187/live/8015/callback?uuid={}&host={}", "livecontribution": "http://114.114.114.114:8187/live/8015/callback?uuid={}&host={}", "livegoods": "http://114.114.114.114:8187/live/8015/callback?uuid={}&host={}", "livesquare": "http://114.114.114.114:8186/livesquare/8015/callback?uuid={}&host={}", "product": "http://114.114.114.114:8180/product/8015/callback?uuid={}&ip={}", "topic": "http://114.114.114.114:8181/topic/8015/callback?uuid={}&host={}", "videogoods": "http://114.114.114.114:8185/videoproduct/8015/callback?uuid={}&host={}", "hourlist": "http://114.114.114.114:8186/liverank/8015/callback?uuid={}&host={}", "videourl": "http://114.114.114.114:8184/videourl/7012/callback?uuid={}&host={}"}
# 当前设备username
username=
# 消息发送地址
inform_url=https://open.feishu.cn/open-apis/bot/v2/hook/1234
# 工作空间标识
uuid=19f97342-fc64-40e3-b9d2-e5a48e0b8549

注意：首先是项目路径“project”，一定配置正确；其次如果你有多个手机的话直接仿造device1的配置项在build.ini创建一个新的配置项，以此类推；然后你想要hook那个app自己补上对应的ui名称和包名，当然有些项是专属的，比如username可以不用、inform_url自己的通知api等等，具体看配置文件的说明；最后一定要把build.ini放在deploy文件夹下，写死的

3.deploy.py的用法

上图中快速生成工作空间代码的使用场景：

• 第一行：linux中一键部署会用到，指令：
  python deploy.py release_workspace
  python deploy.py receive_workspace ， 如果指的收纳的时候要覆盖当前的build.ini文件的配置项，在指令后面加个参数True
• 第二行：本地调试，生成工作空间
• 第三行：本地调试，收纳工作空间
• 第四行：本地调试，生成指定的工作空间
• 第五行：本地调试，收纳指定的工作空间，如果指的收纳的时候要覆盖当前的build.ini文件的配置项，第一个参数True

4.项目启动文件说明

# !/usr/bin/env python
# coding=utf-8
# @Time    : 2022/02/23
# @Author  : XQE
# @Software: PyCharm


"""
整个采集项目的启动入口
兼容windows和linux系统
"""


import os
import configparser


cmd_str = r"""#!/bin/bash

source /etc/profile
echo "============================== Linux deploy start ! =============================="
crontab -l | grep -v '{project_path}/workspaces/monitor' | crontab -
cd {project_path}/startproject
python3.8 /wechatserver/wxhook/startproject/kill_process.py
cd {project_path}/deploy
python3.8 {project_path}/deploy/deploy.py receive_workspace
python3.8 {project_path}/deploy/deploy.py release_workspace
sleep 3
nohup python3.8 {project_path}/startproject/reboot_phone.py >/dev/null 2>&1 &
sleep 120
if [ -d "{project_path}/workspaces/Process" ]; then
    rm -rf {project_path}/workspaces/Process
fi

echo "============================== Linux deploy end ! =============================="
echo "****************************** Start youwant project . ******************************"

cd {project_path}/workspaces/device1
bash {project_path}/workspaces/device1/device1.sh
echo "comment1 start OK ! "

sleep 60
echo "*/5 * * * * . /etc/profile;cd {project_path}/workspaces/monitor && python3.8 {project_path}/workspaces/monitor/monitor.py &" >> /var/spool/cron/root
echo "****************************** Start youwant project OK ! ******************************"
echo "============================== All steps completed ! =============================="
"""


def read_conf(config_path):
    """
    读取监控配置文件
    :param config_path:配置文件路径
    :return:配置文件
    """
    config = None
    try:
        config = configparser.ConfigParser()
        config.read(config_path, encoding="utf-8")
    except Exception as e:
        print("Project is not run .")
    return config


def create_shell(start_sh, project_path):
    """
    创建项目启动文件
    :param start_sh: 项目启动sh文件
    :param project_path: 项目目录
    :return:
    """
    with open(start_sh, mode="w", encoding="utf-8") as f:
        f.write(cmd_str.format(project_path=project_path))


if __name__ == '__main__':
    project_path = read_conf("../deploy/build.ini")["project"]["path"]
    start_sh = os.path.join(os.path.join(project_path, "startproject"), "start_project.sh")
    create_shell(start_sh, project_path)

注意：“cmd_str ”这里面的shell脚本才是关键，其他配置我都写好了，只是当你在build.ini添加了新设备的配置项之后，这边也要加上对应的项目进程启动入口，下面有原本有的device1，假如我加了device2配置项：

• 原有的：

cd {project_path}/workspaces/device1
bash {project_path}/workspaces/device1/device1.sh
echo "device1 start OK !"

• 添加device2:

cd {project_path}/workspaces/device1
bash {project_path}/workspaces/device1/device1.sh
echo "device1 start OK !"

cd {project_path}/workspaces/device2
bash {project_path}/workspaces/device1/device2.sh
echo "device2 start OK !"

这样就每个进程池或者进程都十分明显了

5.本地调试

在本地调试的话，就不需要用到一键部署了，startproject目录下的文件对你来说就是鸡肋，就很简单了：找到deploy文件夹下的deploy.py，参考3、deploy.py的用法，它会根据你的build.ini自动帮你解压出所有的工作空间，类似上面的device1，直接在本地运行即可

6.远程部署（Jenkins）

远程部署需要用到Jenkins，这边不做太多的赘述了主页有搭建步骤。我们利用Jenkins拉去代码到远程Linux服务器，在拉完代码之后执行start_project.py文件，它会在startproject目录下生成一个start_project.sh文件，利用bash运行它即可，用过Jenkins的应该指定怎么配置了把（手动滑稽）

7.路径的问题

会遇到一些路径的问题，一定要记得该项目路径：

• reboot_phone.py
• start_project.py
• build.ini
  尤其是前面两个，不行的话就直接该代码，我用的是路径拼接的，可以直接传完整路径，都是“project_path”这个属性，自行修改

8.源码

github地址
如果觉得有用，拉完就点个赞

总结

以上叭叭叭说了这么多，看到最后的都是牛人，先脸皮厚要个三连吧，不管你看没看懂，创作真的累，分享思想和技术更累，要想着怎么说才能通俗易懂。。。。。。

总结一下吧：文章中提到的技术，在我博客首页都有相关的博客学习，还是那句话：看到就是赚到。架构搭建不是那么麻烦，很多东西都已经封装好了，自己扩展扩展也许就能满足你自己的业务了（哈哈哈哈）动手试试吧。这个代码的思路是我沉淀几个月的时间了，但是我相信还有漏洞或者不完美，欢迎使用后回馈哈，也许你能优化到更加简便，到时候欢迎在评论区留下自己的见解哈。最后关注博主，带你学习一些正常的、奇奇怪怪的技术哈~~~

心情复杂。。。。。。