中间人与自动化
Apr 9, 2021
5 minute read

前言

自动化控制的技术一般是来做 App 测试的,但对于爬虫工程师来说,这项技术也可以用来爬取数据。

正常来说,爬虫工程师开发爬虫,首先要对 App 进行抓包分析,分析其功能所用的服务端接口。 找出调用接口所需要的方式及所支持的参数。之后就可以愉快地构造接口请求,对其服务端接口发起请求,获取我们所需要的数据。

但如今的 App 厂商往往为了避免这些服务端接口被恶意利用,会使用很多种方法或者技术去防止这种情况的发生。

比如在接口层面上,添加签名及时间戳参数。 添加签名参数的原因,一是防止参数被中间人修改,二是防止接口被调用; 那添加时间戳参数就是为了避免同样的参数被反复调用。

或者对请求及响应的内容进行加密。这样抓包看到的都是密文,接口分析工作就无从下手了。

在网络通信应用层面上,可以替换掉通信协议。这样连抓包都不好抓包了,不是熟悉的 HTTP 协议…

这些方法一般都只能通过对 App 进行逆向分析,找出加签加密算法;找出负责网络通信的模块。 再逆向出算法或协议来,或者沙盒去直接调用它。

自动化控制 App 爬取数据

但逆向是真的麻烦…这时候,自动化控制就派上用场了~

本博客以 Android App 为例子来解释说明。但实际上这个方法可以用在各种场景,无论是通过浏览器去访问某个网页,IOS App,甚至是小程序…

如果所需要的数据展示在 Android App 界面上,就可以通过自动化技术控制 App ,再解析界面的 XML 数据 1,直接获取到这些数据。

如果所需要的数据没有展示,自动化控制技术再配合上中间人抓包,就能获取到这些数据了。 这种情况很常见,服务端接口返回数据的结构是展示的超集(为了复用嘛,也不是 App 的服务端接口都会用上 GraphQL 2 的)。

所以直接对 App 做中间人抓包 3 不香吗?为啥要拿它渲染后,展示在屏幕上的的数据。

如果所需要的数据被加密或者 App 用了另外种网络通信协议,这样就需要结合逆向来做。 只需要逆向定位到这一部分的的代码,利用 frida 4 等调试工具(误),获取明文数据。 而不需要完全逆向出整套代码。 但涉及到了这一步,一般都会通过 frida 做成 RPC 服务,去直接调用 App 里对应的函数来获取数据。 (通常是加签函数,然后就可以调用服务端接口了)

实现

现在我展示一下对安卓手机上的 APP 进行抓包,配合自动化控制去爬取数据。 首先要介绍的是怎么对 App 进行中间人抓包。这里以 mitmproxy 5 这个轮子为例子。

中间人代理配置

pip install mitmproxy 安装,再通过 mitmproxy 运行。 然后你就能看到一个交互式终端 UI,这时候你的设备就部署好了一个中间人 MITM 代理服务。

Android 配置

把手机链接到与代理服务一样的 Wi-Fi 下,或者以其他方式链接到能与其通信的网络。 在 WI-FI 配置里配置代理服务,输入代理服务的 ip 地址及端口。 用手机的浏览器打开这个网页链接 http://mitm.it 。 若是配置正确的话,打开后即可看到这段文字 Install mitmproxy’s Certificate Authority。 且终端里的 mitmrpoxy 也能看到该浏览器的流量。 再根据网页的教程安装好证书后,我们就可以进行下一步了。6 (安装证书是为了抓包 HTTPS 的流量 7

由于 7 及 7 以上版本的安卓系统允许 App 不使用用户安装的证书,这样照以上方法安装了证书后,有可能抓不到 App 的网络流量。 得把 mitmproxy 的证书安装成系统自带的证书,才能正常抓包。安装方式可参考 Install System CA Certificate on Android Emulator | mitmproxy docs。(也有其他的方法能解决这个问题)

抓包分析

通过实际操作手机 App 使用特定功能,抓包查看其网络流量,发现其数据来自该 App 服务端的某一个接口。 再确定如何解析这接口所响应的数据,拿到我们想要的内容。 假如这接口的参数有加签,这样我们就需要逆向其加签的算法。 如果在综合考虑下,自动化模拟用户操作抓包爬取数据的方案适合的话,我们就可以通过本文介绍的方式进行数据爬取。

mitmproxy 插件功能

mitmproxy 除了能抓包网络流量,还支持自定义编写插件对流量进行读取,重写等操作。 根据 mitmproxy 提供的插件功能 8 , 9,我们很方便就可以写出,从该 App 的网络流量中获取数据的代码。

import mitmproxy.http
from yarl import URL

class Interceptor:
    def response(self, flow: mitmproxy.http.HTTPFlow):
        respose_url = URL(flow.request.url)
        if respose_url.path != "/feed":  # drop the flows we dont need
            return

        assert flow.response is not None
        response_text = flow.response.text
        # parsing and extracting
        data = json.loads(response_text)
        print(do_extracting(data))

addons = [
    Interceptor()
]

通过执行命令 mitmdump -w traffics.flow 来一边操作 App 使用特定功能,一边录制这过程所产生的网络流量。 再通过命令 mitmdump -s interceptor.py -n -r traffics.flow 重放刚录制到的网络流量,测试以上代码是否能正常的解析获取我们所需要的数据。

当开发完成后,就可以省去这两步。 直接 mitmdump -s interceptor.py 再运行自动化控制脚本去操作 App 使用特定功能,就能自动获取我们想要的数据。

自动化控制

本文以 uiautomator2 这个轮子来实现对 Android App 的自动化控制。 它通过对 UI Automator 进行封装,封装成为一个 RPC 服务,部署在 Android 上。 再通过这个轮子提供的函数库,对这个服务暴露的接口进行调用,达到自动化控制的目的。

import uiautomator2 as u2

d = u2.connect() # connect to device
d.app_start("com.android.XXX")
... # and more example on repo openatx/uiautomator2

这个项目能让我们通过 Python 写代码去自动化控制 Android 设备。 而 IOS 设备的话,虽然我没接触过,但应该也有类似的项目。 浏览器自动化的话,那就多了,像 pyppeteerSeleniumPlaywright。 推荐用 pyppeteer 和 playwright,因为都支持 asyncio。 但又因为这两个项目都挺新的,坑应该不少。 比如 pyppeteer 不能配置 --remote-debuging-port

集成

现在要把这两部分结合起来。一般会想到的方法是,写成两个程序,一个负责抓包,一个负责操作,同时运行。 这种方法虽然实现起来简单,但有很多不足。 比如应对复杂的场景像滑动验证码破解,需要抓包多个接口的响应内容,再执行不同的自动化控制操作。(懒得逆向,就是自动化控制+中间人抓包一把梭哈)

需要从用代码去部署 mitmdump 而不是用 CLI。 这样中间人和自动化控制的代码就可以在同一个进程运行,这样很方便地共享各自的状态,根据这些状态做相应的调整。 我们直接打开 mitmproxy CLI 部分的源码,看看它是怎么部署 mitmproxy 的。 打开 /mitmproxy/tools/[email protected]#L147,从这源码开始看。

mitmproxy.tools.main.run 函数中可以看出,要先创建 mitmproxy.options.Optionsmitmproxy.proxy.ProxyConfig 对象用来传递参数,再创建 mitmproxy.master.Mastermitmproxy.proxy.ProxyServer 对象。 由于底层会用到 mitmproxy.addons.core.Core 所注册的参数 10 , 所以需要添加进来。好像会在下个大版本后删除,但目前就只能加上这行代码了。

你应该还会注意到,以下代码没有用 Master.shutdown 去关闭 MITM 。 因为该方法导致进程不能正常退出,所以用以下两行代码来代替。 11 (但是不懂为啥会出现这个情况,不过应该跟 all non-daemon threads 有关系,得进一步排查看看)

from mitmproxy.addons import core
from mitmproxy.master import Master
from mitmproxy.options import Options
from mitmproxy.proxy import ProxyConfig, ProxyServer


async def main() -> None:
    options = Options(listen_host="0.0.0.0", listen_port=8080)
    master = Master(options)
    master.server = server = ProxyServer(ProxyConfig(options))

    # TODO: Keeps this line until the major version 7 is released
    master.addons.add(core.Core())

    master.start()
    await master.running()
    loop = asyncio.get_event_loop()
    try:
        # Do thing with MITM
        pass
    finally:
        # Don't use `Master.shutdown` method.
        # It will make the process being unable to exit.
        # And I don't know what is causing this.
        master.should_exit.set()
        server.shutdown()


if __name__ == "__main__":
    asyncio.run(main())

用以上代码即可用脚本代码去运行 MITM 。那接下来需要实现的功能就是,能方便的注册及反注册钩子,被不同网络事件(像 TCP 链接,HTTP 请求发出等事件)触发。 12

文档中提到可以注册的钩子有很多,但通常情况下我们只需要的就是 HTTP 响应的钩子。 通过注册自定的 addon 达到注册 HTTP 响应钩子的目的。 但每次都注册都要整个 Addon 类,还是挺麻烦的。 直接一步到位,实现一个可以自由切换钩子的 Addon 类即可。顺便重构一下启动 MITM 的代码,使其支持上下文管理器语法,更易用。

...
import contextlib

...
from mitmproxy.http import HTTPFlow


ResponseHook = Callable[[HTTPFlow], None]


class Interceptor:
    def __init__(self):
        self.response_hook: Optional[ResponseHook] = None

    @contextlib.contextmanager
    def hook_response(self, hook_function: ResponseHook):
        self.response_hook = hook_function
        try:
            yield
        finally:
            self.response_hook = None

    def response(self, flow: HTTPFlow):
        if self.response_hook is not None:
            self.response_hook(flow)


@contextlib.asynccontextmanager
async def intercept(listen_port: int = 8080) -> None:
    options = Options(listen_host="0.0.0.0", listen_port=listen_port)
    master = Master(options)
    master.server = server = ProxyServer(ProxyConfig(options))

    # TODO: Keeps this line until the major version 7 is released
    master.addons.add(core.Core())

    master.start()
    await master.running()

    interceptor = Interceptor()
    master.addons.add(interceptor)
    try:
        yield interceptor
    finally:
        # Don't use `Master.shutdown()`.
        # It will make the process being unable to exit.
        # And I don't know what is causing this.
        master.should_exit.set()
        server.shutdown()

以下是使用例子。 以 httpx 13 HTTP 客户端为例子,通过配置 MITM 为代理,通过对 httpbin.org 做网络请求,就会触发 HTTP 响应钩子。 两份函数做的事情都是一样的,只是演示一下如何处理会堵塞主线程的代码。 由于 control 函数如果直接跑的话,会堵塞住主线程,这样导致了 MITM 不能正常工作。 所以要把这个函数通过 asyncio.to_thread 让该函数在另一个线程去跑,这样才不会阻塞主线程。 而 async_control 函数是协程函数,不会堵塞住主进程的,所以直接跑就好了。

无论是 httpx.Client 还是 httpx.AsyncClient 都是通过相同的参数 proxies 去配置其所使用代理服务。 由于是对 https://httpbin.org 做请求,需要安装证书才能正常处理经过 MITM 的网络流量。 7 只需通过参数 verify 去配置。

from pathlib import Path

import httpx


CAFILE_PATH = Path("~/.mitmproxy/mitmproxy-ca.pem").expanduser().absolute()


def control(interceptor: Interceptor, proxy_port: int):
    def wait_resposne(flow: HTTPFlow):
        # run in main thread
        print("in sync control")
        print("received response url:", flow.request.url)

    with httpx.Client(
        proxies=f"http://localhost:{proxy_port}", verify=CAFILE_PATH
    ) as client:
        # run in another thread
        with interceptor.hook_response(wait_resposne):
            resp = client.get("https://httpbin.org/get")
            resp.raise_for_status()

        resp = client.get("https://httpbin.org/get")
        resp.raise_for_status()


async def async_control(interceptor: Interceptor, proxy_port: int):
    def wait_resposne(flow: HTTPFlow):
        print("in async control")
        print("received response url:", flow.request.url)

    async with httpx.AsyncClient(
        proxies=f"http://localhost:{proxy_port}",
        verify=CAFILE_PATH,
    ) as client:
        with interceptor.hook_response(wait_resposne):
            resp = await client.get("https://httpbin.org/get")
            resp.raise_for_status()

        resp = await client.get("https://httpbin.org/get")
        resp.raise_for_status()


async def main():
    loop = asyncio.get_event_loop()
    port = 8888
    async with intercept(listen_port=port) as interceptor:
        # run in other thread to avoid blocking main thread
        await loop.run_in_executor(None, control, interceptor, port)

        # async function is invoked normally via await statement
        await async_control(interceptor, port)


if __name__ == "__main__":
    asyncio.run(main())

查看运行的结果,跟我们预料的一样,两个函数各做了两次请求,不过都只触发了一次钩子。

in sync control
received response url: http://httpbin.org/get
in async control
received response url: http://httpbin.org/get

以上代码片段的完整版: Gist

中间人与自动化

现在结合 uiautomator2 和一台 Android 设备,无论是虚拟机,云手机还是实体机,都可以通过以下方式中间人获取 App 的网络数据。

def configure_remote_proxy(proxy_port):
    raise NotImplementedError


def control(interceptor):
    raise NotImplementedError


async def main():
    port = 8888
    async with intercept(listen_port=port) as interceptor:
        # Run automation script to configure proxy setting of remote device.
        await asyncio.to_thread(configure_remote_proxy, proxy_port=port)
        # Due to the uiautomator2 does not support asyncio
        await asyncio.to_thread(control, interceptor=interceptor)



if __name__ == "__main__":
    asyncio.run(main())

配置代理的方式有很多,可以参考以下链接。

只要把这个配置代理过程做成自动化就好了,具体情况具体实现。 再配合 interceptor.hook_response 方法去实现一边自动化控制,一边中间人获取数据,达到我们的最终目的。

无论你是做测试还是做爬虫,我相信这篇博客都能给你带来一点启发。

Footnotes

1 布局定义 | Android Docs

2 GraphQL | A query language for your API

3 中间人攻击

4 Frida • A world-class dynamic instrumentation framework

5 mitmproxy - an interactive HTTPS proxy

6 Certificates | mitmproxy docs

7 How Does SSL Work? | SSL Certificates and TLS

8 Addons | mitmproxy docs

9 Events | mitmproxy docs

10 Addon 通过实现 load 方法,注册自动处理参数的函数(CLI 参数处理,设置文件处理)。 详情见 /mitmproxy/addons/[email protected]#L22-L36

11 Master.shutdown 方法的代码 /mitmproxy/[email protected]#L117-L131

12 mitmproxy 脚本例子,如何利用这些钩子 /mitmproxy/[email protected]

13 HTTPX | A next-generation HTTP client for Python.




comments powered by Disqus