碧蓝航线实时投票排行榜与数据抓取

@Pelom  December 18, 2021
技术 

前言

碧蓝航线的人气投票今年已经是第三届了，在b站看到过不少实时排行榜的视频，觉得这种东西很有趣，决定给碧蓝也做一个。开始动手做的时候才发现许多相关的东西都不会，比如获取数据，制作图表，服务器部署等。不会就得学，还好在决赛之前把网站部署好了

抓取数据

历程

  1. 首先想到能不登游戏直接获取肯定是最好的，于是开始研究fiddler。可是登入游戏之后就只有与b站的心跳包，于是上网搜索解决方法。途中了解了游戏为防止中间人攻击而使游戏数据的传输不经过代理，以及高版本安卓中的软件可以选择自己信任的证书。解决了这些问题之后，仍然抓不到数据包，只好就此作罢
  2. 既然不能直接抓包，只好换条路。这次选择的办法是在云手机上进行OCR。等到OCR处理好之后才想起网易云游戏上面没有碧蓝航线，并且通过自动化执行OCR很麻烦，情况很复杂，于是又放弃了这条路
  3. 于是又回到抓包这条路，这次换了wireshark，由于其工作原理与fiddler不同（wireshark基于网卡复制），决定再尝试一次。打开wireshark就是满屏滚动的数据，没办法，只好一点一点过滤。最后发现使用的是socket通信

分析

结合github上的碧蓝航线lua脚本，可以得出客户端和服务器通信的流程

编码方式

首先要了解发送信息的编码方式，通过lua脚本中管理连接的部分得知，信息的结构为（以下均为hex编码）
$$\mathbf{\underbrace{000a}_{protobuf部分长度}\ \underbrace{00}_{占位0}\ \underbrace{2a30}_{指令号码}\ \underbrace{0000}_{指令时间戳}\ \underbrace{082f120130}_{protobuf部分}}$$
例如以上这段数据中，$\mathbf{000a}$表示最后一部分的长度为10，$\mathbf{2a30}$表示某种指令，$\mathbf{0000}$表示这是第1条指令，$\mathbf{082f120130}$为protobuf编码，接下来会讲到

指令号码

关于指令，在游戏中按住右侧同时在左侧滑动，可以开启日志界面。其中有每一次与服务器通信的log，会输出指令号码，有助于在脚本中定位到相关逻辑
例如拉取投票信息的指令号码为17203，转为hex后为$\mathbf{4333}$，恰好与wireshark抓到的一致

指令时间戳

一个递增的数，用于区分指令

ProtoBuf

关于protobuf，是Google提出的一种数据编码方式，其在效率，兼容性方面均远远优于传统数据交换常用的JSON和XML。但这并不意味着它没有损失，作为代价，它失去了自描述性。如果没有配套的.proto文件，我们将无法得知它描述了什么内容
这里只解释用到的部分

  1. Wire Type
    每一项数据之前都会有一个tag，表示对应的数据类型$$\text{tag=(field\_number << 3) | wire\_type}$$其中field_number用以给数据标号，wire_type则表示了数据类型。例如0代表varint，2代表string
  2. Varint
    varint是一种编码方式，能使不同量级的数以不同的空间存储。一个varint的每个字节的最高位表示下一个字节是否还属于这个数（0表示结束，1表示继续），并且采用小端的表示方法（即低地址对应低位）。
    编码时，将原数分为7位一组，高位不足补0。如$114514_{(10)}$直接写成二进制为$00011011111101010010_{(2)}$，而使用protobuf的编码过程为
    $$\begin{aligned} \to &{\color{red}0}000110\ 1111110\ 1010010 \\ \to &1010010\ 1111110\ 0000110 \\ \to &{\color{red}1}1010010\ {\color{red}1}1111110\ {\color{red}0}0000110 \end{aligned}$$
    再转为hex即为$\mathbf{d2fe06}$
  3. String
    存储string时，由一个紧跟在tag后面的varint声明string的长度，其后是string的utf-8编码

通信流程

要想抓取投票信息要先登录，否则服务器断开连接
登录流程为：启动程序后，客户端首先向b站服务器发送用户token（不变），随后b站服务器发回一个登录游戏用的token（每次更改），客户端再将包含这个token的信息发给游戏服务器，完成登录
随后发送获取投票信息的指令即可

登录

  1. b站服务器ip为118.178.152.242，端口为80
    token需要自行用wireshark抓包获取，或重发数据流亦可
    返回的token共42位（hex），在整个数据流的[-44:-2]

  2. 游戏服务器ip为203.107.54.123，端口与区服有关，通过http请求http://203.107.54.123/?cmd=load_server?可以获取到

    [{"id":3,"name":"霸王行动","state":0,"flag":0,"sort":3},{"id":4,"name":"冰山行动","state":0,"flag":0,"sort":4},{"id":1,"name":"莱茵演习","state":3,"flag":0,"sort":1},{"id":5,"name":"彩虹计划","state":0,"flag":0,"sort":5},{"id":12,"name":"开罗宣言","state":0,"flag":0,"sort":12},{"id":6,"name":"发电机计划","state":0,"flag":0,"sort":6},{"id":8,"name":"十字路口行动","state":0,"flag":0,"sort":8},{"id":19,"name":"八月风暴","state":0,"flag":0,"sort":19},{"id":17,"name":"瓦尔基里行动","state":0,"flag":0,"sort":17},{"id":16,"name":"白色方案","state":0,"flag":0,"sort":16},{"id":7,"name":"瞭望台行动","state":0,"flag":0,"sort":7},{"id":11,"name":"地狱犬行动","state":0,"flag":0,"sort":11},{"id":9,"name":"朱诺行动","state":0,"flag":0,"sort":0},{"id":18,"name":"曼哈顿计划","state":0,"flag":0,"sort":18},{"id":14,"name":"小王冠行动","state":0,"flag":0,"sort":14},{"id":13,"name":"奥林匹克行动","state":0,"flag":0,"sort":0},{"id":15,"name":"波茨坦公告","state":0,"flag":0,"sort":0},{"id":10,"name":"杜立特空袭","state":0,"flag":0,"sort":0},{"id":2,"name":"巴巴罗萨","state":3,"flag":0,"sort":2},{"id":20,"name":"秋季旅行","state":3,"flag":0,"sort":20},{"id":21,"name":"水星行动","state":0,"flag":0,"sort":21}]

    端口为80+id。例如奥林匹克行动，端口即为8013
    token在发往游戏服务器之前会再算出一个MD5值作为验证，从脚本中查得其算法为

    salt: str = 'dettimrepsignihtyrevednaeurtsignihton'
check: bytes = md5(token + salt.encode("utf8")).hexdigest().encode("utf8")

    整个数据流为

    data: bytes = b'\x00\x5f\x00\x27\x26\x00\x00\x08' + uid + '\x12\x2a' + token + b'\x1a\x01\x30\x20\x0e\x2a\x20' + check + b'\x32\x00'

    该数据流中包含平台、uid等信息，有可能与上例不同，需要自行用wireshark抓取，可对比替换token、check的部分
    随后发往游戏服务器进行登录

获取投票信息

前面提到每个数据流中有一个递增的时间戳，即为此处的idx

idx: bytes = bytes([id // 256, id % 256])
data: bytes = b'\x00\x07\x00\x43\x33' + idx + b'\x08\x1b'
s.send(data)
res = s.recv(2048).hex()

对res解码可得投票信息
针对只有数的投票信息，写了个简单的解码函数

def decode(s: str) -> list:
    s = [s[i * 2 : i * 2 + 2] for i in range(len(s) // 2)][7:]
    res: list = []
    f: bool = False
    for i in s:
        i = int(i, 16)
        if f == 0:
            f = True
            num: int = 0
            cnt: int = 0
        else:
            if i >> 7:
                f = True
                num = num | (i & 127) << cnt
                cnt = cnt + 7
            else:
                f = False
                num = num | (i & 127) << cnt
                cnt = cnt + 7
                res.append(num)
    res = [res[i * 5 + 1 : i * 5 + 4 : 2] for i in range(len(res) // 5)]
    return res

数据可视化

使用了echarts.js
基本上是照搬官网示例，不过因为获取到的信息是角色编号，需要做一步转换，里面还有和谐名用代码表示，所幸脚本里面都找得到

服务器部署

vote.py总体代码如下

import socket
import time
import json
from hashlib import md5

salt = 'dettimrepsignihtyrevednaeurtsignihton'
info: list

def decode(s: str) -> list:
    s = [s[i * 2 : i * 2 + 2] for i in range(len(s) // 2)][7:]
    res: list = []
    f: bool = False
    for i in s:
        i = int(i, 16)
        if f == 0:
            f = True
            num: int = 0
            cnt: int = 0
        else:
            if i >> 7:
                f = True
                num = num | (i & 127) << cnt
                cnt = cnt + 7
            else:
                f = False
                num = num | (i & 127) << cnt
                cnt = cnt + 7
                res.append(num)
    res = [res[i * 5 + 1 : i * 5 + 4 : 2] for i in range(len(res) // 5)]
    return res[:32]

def connect() -> None:
    print("connecting...")

    data: bytes = b'\x00\x0a\x00\x2a\x30\x00\x00\x08\x2f\x12\x01\x30'
    s1.send(data)
    s1.recv(1024)
    time.sleep(1)
    s1.send(data)
    s1.recv(1024)
    time.sleep(1)

    # copy your token to here
    data: bytes = b''
    
    s2.send(data)
    token = s2.recv(2048)[-44:-2]
    time.sleep(1)

    check = md5(token + salt.encode("utf8")).hexdigest().encode("utf8")

    # copy your uid(hex) to here 
    data: bytes = b'\x00\x5f\x00\x27\x26\x00\x00\x08' + uid + '\x12\x2a' + token + b'\x1a\x01\x30\x20\x0e\x2a\x20' + check + b'\x32\x00'
    s3.send(data)
    s3.recv(1024)
    time.sleep(1)

    print("connected.")

def pull(id: int) -> str:
    idx = bytes([id // 256, id % 256])
        
    data: bytes = b'\x00\x07\x00\x43\x33' + idx + b'\x08\x1b'
    s3.send(data)
    res = s3.recv(2048).hex()
    if res[6:10] == "4334":
        return res
    else:
        return pull(id)

if __name__ == "__main__":
    with open("./data/vote.json", "r", encoding="utf-8") as f:
        voteJson = json.load(f)

    s1 = socket.socket()
    s1.connect(("203.107.54.123", 80))
    s2 = socket.socket()
    s2.connect(("118.178.152.242", 80))
    s3 = socket.socket()
    # modify port to yours
    s3.connect(("203.107.54.123", 8013))
    
    connect()

    while True:
        for id in range(1, 65536):
            infos = pull(id)
            infos = decode(infos)
            with open("./data/info.json", "w", encoding="utf8") as f:
                json.dump(infos, f, separators=(",", ":"))

            current = time.localtime()
            # save per hour
            if current.tm_min == 0:
                current = time.strftime("%Y-%m-%d %H:%M:00", current)
                for info in infos:
                    info.append(current)
                    voteJson.append(info)
                with open("./data/vote.json", "w", encoding="utf-8") as f:
                    json.dump(voteJson, f, separators=(",", ":"))
                print("save:", id, current)

            # pull per minute
            time.sleep(60)

因为是python3写的抓取程序，所以要给服务器装python3
起初我担心异地登录的验证码过不了，谁知道它这个验证码好像只是客户端的（笑

运行

命令行执行

nohup python3 vote.py -u > nohup.out 2>&1 &

表示不挂断地执行（即关闭终端后继续执行），并将输出到重定向到nohup.out

中止

命令行执行

ps -aux

显示所有进程，并找到python3 vote.py的pid，之后执行

kill -9 pid

网站

后话

其实按这个道理的话也可以进行签到等活动（当然你要是不怕被官方查水表的话）

引用

添加新评论