GZCTF平台对接QQ机器人

本文章写于2024-04-23，不保证后续可用性。

参考

https://github.com/mamoe/miraihttps://github.com/mamoe/mirai
https://qchatgpt.rockchin.top/posts/deploy/platforms/mirai.html#安装-mirai-api-httphttps://qchatgpt.rockchin.top/posts/deploy/platforms/mirai.html#安装-mirai-api-http

采用mirai项目

安装mirai

用mirai-installer安装

https://github.com/iTXTech/mcl-installerhttps://github.com/iTXTech/mcl-installer

安装

1
./mcl-installer

初始化

1
./mcl

安装完毕

插件安装及配置

mirai-http-api插件用于开放接口

https://github.com/project-mirai/mirai-api-httphttps://github.com/project-mirai/mirai-api-http

打开mcl后输入

1
mcl --update-package net.mamoe:mirai-api-http --channel stable-v2 --type plugin

即可安装成功

配置文件setting.yml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
## 配置文件中的值，全为默认值

## 启用的 adapter, 内置有 http, ws, reverse-ws, webhook
adapters:
   - http 
  - ws

## 是否开启认证流程, 若为 true 则建立连接时需要验证 verifyKey
## 建议公网连接时开启
enableVerify: true
 verifyKey: 1234567890
 
## 开启一些调试信息
debug: false

## 是否开启单 session 模式, 若为 true，则自动创建 session 绑定 console 中登录的 bot
## 开启后，接口中任何 sessionKey 不需要传递参数
## 若 console 中有多个 bot 登录，则行为未定义
## 确保 console 中只有一个 bot 登录时启用
singleMode: false

## 历史消息的缓存大小
## 同时，也是 http adapter 的消息队列容量
cacheSize: 4096

## adapter 的单独配置，键名与 adapters 项配置相同
adapterSettings:
  ## 详情看 http adapter 使用说明 配置
  http:
     host: localhost
    port: 8080 
    cors: ["*"]
    unreadQueueMaxSize: 100
  
  ## 详情看 websocket adapter 使用说明 配置
  ws:
    host: localhost
    port: 8080
    reservedSyncId: -1

采用http，verifykey是后续用于认证的。如若使用Docker开放端口，host要改为0.0.0.0，不然访问不到。

我的配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
adapters:
  - http

enableVerify: true
verifyKey: 1234567890

debug: false
singleMode: false
cacheSize: 4096

adapterSettings:
  http:
    host: 0.0.0.0
    port: 7777
    cors: ["*"]
    unreadQueueMaxSize: 100

loginsolver插件用于登录，拖入plugins即可

https://github.com/KasukuSakura/mirai-login-solver-sakurahttps://github.com/KasukuSakura/mirai-login-solver-sakura

需要用到22333端口，docker中记得开放，登录时会返回地址，需要用手机下载Sakuralogin来输入地址登录。

签名服务器Qsign

https://github.com/MrXiaoM/qsignhttps://github.com/MrXiaoM/qsign

下载时下载d62ddce版本，将plugins中的jar放到bot的plugins中，将txlib放到bot的根目录即可。

要修改配置文件中的协议版本为8.9.90

配置即可完成

运行

运行mcl

1
./mcl

使用内置命令登录，最好使用ANDROID_PAD，手表协议发20条就会风控。

1
/login qq password ANDROID_PAD

登录完之后可以设置autoLogin，具体不赘述。

使用

使用的什么adapter就看什么，详细描述了api接口

https://docs.mirai.mamoe.net/mirai-api-http/adapter/HttpAdapter.htmlhttps://docs.mirai.mamoe.net/mirai-api-http/adapter/HttpAdapter.html

简单实现的一个python bot类，抛砖引玉，还可以继续扩展。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
import requests

VERIFY_KEY = ''
BOT_URL = 'http://x.x.x.x:7777'


class Bot:
    def __init__(self, verify_key: str, url: str, qq: str):
        self.verify_key = verify_key
        self.url = url
        self.qq = qq
        self.session = self.verify()
        self.bind()

    def verify(self):
        url = f'{self.url}/verify'
        data = {"verifyKey": self.verify_key}
        req = requests.post(url, json=data)
        return req.json()['session']

    def bind(self):
        url = f'{self.url}/bind'
        data = {"sessionKey": self.session, 'qq': self.qq}
        req = requests.post(url, json=data)
        return req.json()['msg']

    def rebind(self):  # 由于30分钟不使用就会过期，所以要重新绑定。
        self.session = self.verify()
        self.bind()

    def get_group_list(self):
        self.check_session()
        url = f'{self.url}/groupList?sessionKey={self.session}'
        rep = requests.get(url)
        return rep.json()

    def check_session(self):  # 检测session是否过期
        url = f'{self.url}/sessionInfo?sessionKey={self.session}'
        rep = requests.get(url)
        if rep.json()['code'] == 3:
            self.rebind()
        return

    def send_group_message(self, id: int, message: str):
        self.check_session()
        url = f'{self.url}/sendGroupMessage'
        data = {"sessionKey": self.session, 'target': id, 'messageChain': [
            {"type": "Plain", "text": message},
        ]}
        req = requests.post(url, json=data)
        return req.json()['msg']


if __name__ == '__main__':
    bot = Bot(VERIFY_KEY, BOT_URL, '0000000000')
    print(bot.get_group_list())
    print(bot.send_group_message(0000000000, '123123'))

对接GZCTF平台

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
import time
from datetime import timezone, timedelta
import requests
from dateutil.parser import isoparse
from bot import Bot
import logging
from rich.logging import RichHandler
from rich.status import Status

INTERVAL = 2  # 间隔时间
FORMAT = "%(message)s"
VERIFY_KEY = ''  # mirai-http-api verify key
BOT_URL = 'http://0.0.0.0:7777'  # mirai-http-api bot url
BOT_QQ = ''  # bot qq号
GROUP_NOTICE_ID = ''  # 发送的群号
BOT = Bot(VERIFY_KEY, BOT_URL, BOT_QQ)
API_URL = 'http://0.0.0.0/api/game/4/notices'  # 比赛通知api
BANNER = ''
NOW_ID = 0
TEMPLATES = {
    'Normal': '【比赛公告】\n内容：%s\n时间：%s',
    'NewChallenge': '【新增题目】\n[%s]\n时间：%s',
    'NewHint': '【题目提示】\n[%s]有新提示，请注意查收\n时间：%s',
    'FirstBlood': '【一血播报】\n恭喜%s拿下[%s]一血\n时间：%s',
    'SecondBlood': ' 【二血播报】\n恭喜%s拿下[%s]二血\n时间：%s',
    'ThirdBlood': ' 【三血播报】\n恭喜%s拿下[%s]三血\n时间：%s'
}


def processTime(t):
    t_truncated = t[:26] + t[26:].split('+')[0]
    input_time = isoparse(t_truncated)
    input_time_utc = input_time.replace(tzinfo=timezone.utc)
    beijing_timezone = timezone(timedelta(hours=8))
    beijing_time = input_time_utc.astimezone(beijing_timezone)
    return beijing_time.strftime("%Y-%m-%d %H:%M:%S")


if __name__ == '__main__':
    logging.basicConfig(
        level=logging.INFO, format=FORMAT, datefmt="[%X]", handlers=[RichHandler()]
    )
    log = logging.getLogger("rich")
    notices = requests.get(API_URL).json()
    notices = sorted(notices, key=lambda x: x['id'])
    NOW_ID = notices[-1]['id']
    #NOW_ID -= 1
    status = Status('Waiting for new notice')
    status.start()
    while True:
        try:
            notices = requests.get(API_URL).json()
        except KeyboardInterrupt:
            log.info('Exit bot')
            break
        except Exception:
            log.warning('Warning: request failed')
            continue
        notices = sorted(notices, key=lambda x: x['id'])
        for notice in notices:
            if notice['id'] > NOW_ID:
                message = TEMPLATES[notice['type']] % tuple(notice['values'] + [processTime(notice['time'])])
                log.info(f'sending to {GROUP_NOTICE_ID} message: \n{message}')
                BOT.send_group_message(GROUP_NOTICE_ID, message)
                NOW_ID = notice['id']
        try:
            time.sleep(INTERVAL)
        except KeyboardInterrupt:
            log.info('Exit bot')
            break
    status.stop()

还是挺稳的。

许可协议

本文采用 署名-非商业性使用-相同方式共享 4.0 国际 许可协议，转载请注明出处。