前言

在搭建了CTFd平台后，需要在QQ群里加一个会自动播报的bot，在网上查阅了相关资料并尝试后发现当前很多框架都用不了了，比如Lagrange.Onebot，go-cqhttp，都是过不了签名。所以我最终选择的方案是NapCat + OneBot 11

napcat部署

这里可以直接参看napcat的官方文档，写得很详细了https://napneko.github.io/ 我这里选择的方式是docker部署

1
sudo docker pull mlikiowa/napcat-docker:latest

安装好之后运行

1
2
3
4
5
6
7
8
9
10
sudo docker run -d \
  -e ACCOUNT=123456789 \
  -e WS_ENABLE=true \
  -e NAPCAT_GID=0 \
  -e NAPCAT_UID=0 \
  -p 3001:3001 \
  -p 6099:6099 \
  --name napcat \
  --restart=always \
  mlikiowa/napcat-docker:latest

然后sudo docker ps来查看是否启动成功

成功之后就继续sudo docker logs napcat

在日志中可以直接扫码登录QQ(登录准备的bot的账号)也可以查看webUI的token来登录web管理页面

webUI的端口是6099，这里要访问的话，如果你开放了端口的那直接公网ip就行，不然就通过ssh

1
ssh -L 6099:127.0.0.1:6099 username@你的服务器公网IP

之后通过之前获取的token登录

登录后在网络配置中进行配置新建http服务器

http_server

这里的sever token要保存下来

查看你对应的CTFd的网络和napcat网络

sudo docker network ls

然后将Napcat接入CTFd的网络

sudo docker network connect aegis_default napcat

检查是否接入

1
2
sudo docker inspect napcat \
  --format '{{range $k,$v := .NetworkSettings.Networks}}{{$k}}{{"\n"}}{{end}}'

看到类似

1
2
bridge
aegis_default

就可以了

因为它们在同一个 Docker 网络中，所以 CTFd 不需要通过宿主机 IP 访问 NapCat，可以直接使用：

1
napcat:3000

同理 NapCat 可以使用：

1
aegis-ctfd-1:8000

访问 CTFd。

这里我还在webUI创建了一个http client server，因为我这里还想实现机器人自动欢迎新成员和查看排名，如果不要这功能的话可以不管这部分

http_client

这里我是直接url明文传送了，因为不知道为什么CTFd的CSRF跨域不让我通过，事已至此先能用再说(ˇзˇ)

CTFd文件修改

general.html

改这个主要是想直接通过前端修改bot的一些简单配置，就是CTFd后台的general页面

该文件存在路径CTFd/themes/admin/templates/configs/general.html

可能会因为版本问题略有不同

加入：

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
<div class="form-group">
    <label for="bot">QQ机器人播报</label>

    <div class="form-check">
        <input class="form-check-input"
               type="checkbox"
               name="bot"
               id="bot"
               value="1"
               {% if bot %}
               checked
               {% endif %}>

        <label class="form-check-label" for="bot">
            启用QQ机器人播报
        </label>
    </div>
</div>

<div class="form-group">
    <label for="bot_ip">NapCat 地址</label>

    <input class="form-control"
           type="text"
           name="bot_ip"
           id="bot_ip"
           value="{{ bot_ip or '' }}"
           placeholder="napcat:3000">
</div>

<div class="form-group">
    <label for="group_id">QQ群号</label>

    <input class="form-control"
           type="text"
           name="group_id"
           id="group_id"
           value="{{ group_id or '' }}"
           placeholder="你的QQ群号">
</div>

<div class="form-group">
    <label for="bottext">播报模板</label>

    <input class="form-control"
           type="text"
           name="bottext"
           id="bottext"
           value="{{ bottext or '' }}"
           placeholder="🎉 恭喜 %s 解出题目「%s」!">
</div>

位置就是接着最后一个form后

docker-compose.yml

位置在/docker-compose.yml根目录下

我这里将之前http server的token写入到了环境中，在里面的environment里面添加NAPCAT_TOKEN: "你的server Token"

现在可以进行测试一下CTFd能否正常访问napcat，因为默认的容器内是没有curl的，所以直接用python脚本（其实是懒得配，反正ai生成检查脚本）(≧▽≦)

执行：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
sudo docker exec -i 你的ctfd容器名称 python - <<'PY'
import os
import urllib.request

token = os.getenv("NAPCAT_TOKEN")

req = urllib.request.Request(
    "http://napcat:3000/get_login_info",
    headers={
        "Authorization": f"Bearer {token}"
    }
)

try:
    print(
        urllib.request.urlopen(
            req,
            timeout=5
        ).read().decode()
    )
except Exception as e:
    print(type(e).__name__, str(e))
PY

正常会返回：

1
2
3
4
5
6
7
{
    "status": "ok",
    "retcode": 0,
    "data": {
        ...
    }
}

这说明：

1
2
3
4
CTFd → NapCat Docker网络正常
NapCat HTTP Server正常
Token正常
QQ登录正常

如果前面配置了http client的话也在这里的environment里面填上client的tokenEVENT_TOKEN: "你的client token"

challenges.py

位置在

1
/CTFd/api/v1/challenges.py

顶部引入

1
2
3
4
import os
import requests

from flask import current_app

在成功解题逻辑附近，就是if status == "correct" or status is True:中，**chal_class.solve()**调用前：

1
2
user_name = user.name
challenge_name = challenge.name

我的 CTFd 注册时用户邮箱是使用 QQ 邮箱，因此尝试提取 QQ：

1
2
3
4
5
6
7
qq_number = None

if user.email and user.email.endswith("@qq.com"):
    candidate = user.email.split("@")[0]

    if candidate.isdigit():
        qq_number = candidate

这个可以紧接着上段代码放

例如：

1
123456789@qq.com

会得到：

1
123456789

之后就是一血、二血、三血播报

完成 solve 后查询当前题目的 solve 数：

1
2
3
solve_count = Solves.query.filter_by(
    challenge_id=challenge_id
).count()

然后：

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
if solve_count == 1:
    botmessage = (
        f"🥇 恭喜 {user_name} "
        f"斩获「{challenge_name}」一血！"
    )

elif solve_count == 2:
    botmessage = (
        f"🥈 恭喜 {user_name} "
        f"拿下「{challenge_name}」二血！"
    )

elif solve_count == 3:
    botmessage = (
        f"🥉 恭喜 {user_name} "
        f"获得「{challenge_name}」三血！"
    )

else:
    if bottext:
        botmessage = bottext % (
            user_name,
            challenge_name
        )
    else:
        botmessage = (
            f"🎉 恭喜 {user_name} "
            f"解出题目「{challenge_name}」！"
        )

最终完整发送逻辑：

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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
bot = get_config("bot")

if str(bot) == "1":

    bottext = get_config("bottext")
    group_id = get_config("group_id")
    bot_ip = get_config("bot_ip")

    bot_token = os.getenv("NAPCAT_TOKEN")

    if group_id and bot_ip:

        try:

            solve_count = Solves.query.filter_by(
                challenge_id=challenge_id
            ).count()

            if solve_count == 1:

                botmessage = (
                    f"🥇 恭喜 {user_name} "
                    f"斩获「{challenge_name}」一血！"
                )

            elif solve_count == 2:

                botmessage = (
                    f"🥈 恭喜 {user_name} "
                    f"拿下「{challenge_name}」二血！"
                )

            elif solve_count == 3:

                botmessage = (
                    f"🥉 恭喜 {user_name} "
                    f"获得「{challenge_name}」三血！"
                )

            else:

                if bottext:

                    botmessage = bottext % (
                        user_name,
                        challenge_name
                    )

                else:

                    botmessage = (
                        f"🎉 恭喜 {user_name} "
                        f"解出题目「{challenge_name}」！"
                    )

            url = (
                f"http://{bot_ip}/send_group_msg"
            )

            payload = {
                "group_id": int(group_id),
                "message": botmessage
            }

            headers = {}

            if bot_token:

                headers[
                    "Authorization"
                ] = f"Bearer {bot_token}"

            response = requests.post(
                url,
                json=payload,
                headers=headers,
                timeout=5
            )

            if not response.ok:

                current_app.logger.warning(
                    "NapCat broadcast failed: %s %s",
                    response.status_code,
                    response.text
                )

        except Exception:

            current_app.logger.exception(
                "NapCat broadcast failed"
            )

这段代码的位置是在

1
2
3
clear_standings()

clear_challenges()

之后就放，依旧在解题成功的逻辑中

然后我还希望有发布新题自动通知的功能，但我这里只有把题目的状态从hidden变成visible才行

找到Patch函数

将它进行修改

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
def patch(self, challenge_id):
    data = request.get_json()

    # Load data through schema for validation but not for insertion
    schema = ChallengeSchema()
    response = schema.load(data)
    if response.errors:
        return {"success": False, "errors": response.errors}, 400

    challenge = Challenges.query.filter_by(id=challenge_id).first_or_404()
    old_state = challenge.state
    challenge_class = get_chal_class(challenge.type)

    try:
        challenge = challenge_class.update(challenge, request)
    except ChallengeUpdateException as e:
        return {"success": False, "errors": {"": [str(e)]}}, 500

    if old_state == "hidden" and challenge.state == "visible":
        bot = get_config("bot")

        if str(bot) == "1":
            group_id = get_config("group_id")
            bot_ip = get_config("bot_ip")
            bot_token = os.getenv("NAPCAT_TOKEN")

            if group_id and bot_ip:
                try:
                    message = (
                        f"📢 新题上架！\n"
                        f"题目：{challenge.name}\n"
                        f"分类：{challenge.category}\n"
                        f"快来挑战吧！"
                    )

                    url = f"http://{bot_ip}/send_group_msg"

                    headers = {}

                    if bot_token:
                        headers["Authorization"] = f"Bearer {bot_token}"

                    response_bot = requests.post(
                        url,
                        json={
                            "group_id": int(group_id),
                            "message": message
                        },
                        headers=headers,
                        timeout=5
                    )

                    if not response_bot.ok:
                        current_app.logger.warning(
                            "NapCat challenge publish failed: %s %s",
                            response_bot.status_code,
                            response_bot.text
                        )

                except Exception:
                    current_app.logger.exception(
                        "NapCat challenge publish broadcast failed"
                    )
    
    response = challenge_class.read(challenge)

    clear_standings()
    clear_challenges()

    return {"success": True, "data": response}

views.py

位置在/CTFd/views.py

在头部确保

1
2
import os
import requests

先在setup()函数的general部分后加上以下内容

1
2
3
4
5
6
7
8
9
10
# QQ Robot
bot = request.form.get("bot")
bottext = request.form.get("bottext")
group_id = request.form.get("group_id")
bot_ip = request.form.get("bot_ip")

set_config("bot", bot)
set_config("bottext", bottext)
set_config("group_id", group_id)
set_config("bot_ip", bot_ip)

效果大致如下

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
def setup():
    errors = get_errors()
    if not config.is_setup():
        if not session.get("nonce"):
            session["nonce"] = generate_nonce()
        if request.method == "POST":
            # General
            ctf_name = request.form.get("ctf_name")
            ctf_description = request.form.get("ctf_description")
            user_mode = request.form.get("user_mode", USERS_MODE)
            set_config("ctf_name", ctf_name)
            set_config("ctf_description", ctf_description)
            set_config("user_mode", user_mode)

            #Robots
            bot = request.form.get("bot")
            bottext = request.form.get("bottext")
            group_id = request.form.get("group_id")
            bot_ip = request.form.get("bot_ip")

            set_config("bot", 1 if bot else 0)
            set_config("bottext", bottext)
            set_config("group_id", group_id)
            set_config("bot_ip", bot_ip)

            # Settings
            challenge_visibility = ChallengeVisibilityTypes(
                request.form.get(
                    "challenge_visibility", default=ChallengeVisibilityTypes.PRIVATE
                )
            )
            ......

这里是对应的前面改的general.html的部分的后端

这里我想实现一个功能就是能@机器人 #rank然后机器人能返回排名,还有就是自动欢迎新成员

在头部引入

1
2
from CTFd.utils.scores import get_standings
from CTFd.plugins import bypass_csrf_protection

在文件末尾添加

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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def send_napcat_group_message(group_id, message):
    bot_ip = get_config("bot_ip")
    bot_token = os.getenv("NAPCAT_TOKEN")

    if not bot_ip:
        app.logger.warning(
            "NapCat bot_ip is not configured"
        )
        return False

    headers = {}

    if bot_token:
        headers["Authorization"] = f"Bearer {bot_token}"

    try:
        response = requests.post(
            f"http://{bot_ip}/send_group_msg",
            json={
                "group_id": int(group_id),
                "message": message
            },
            headers=headers,
            timeout=5
        )

        if not response.ok:
            app.logger.warning(
                "NapCat send failed: %s %s",
                response.status_code,
                response.text
            )

            return False

        return True

    except Exception:
        app.logger.exception(
            "NapCat send group message failed"
        )

        return False

def build_rank_message(limit=10):
    try:
        standings = get_standings()

        if not standings:
            return "🏆 当前排行榜暂无数据"

        lines = [
            "🏆 CTF 个人排行榜 TOP 10",
            "━━━━━━━━━━━━━━"
        ]

        medals = {
            1: "🥇",
            2: "🥈",
            3: "🥉"
        }

        for position, entry in enumerate(standings[:limit], start=1):
            medal = medals.get(position, f"{position}.")

            username = entry.name
            score = int(entry.score)

            lines.append(
                f"{medal} {username}  {score} pts"
            )

        lines.append("━━━━━━━━━━━━━━")

        return "\n".join(lines)

    except Exception:
        app.logger.exception(
            "Failed to build ranking"
        )
        return "⚠️ 获取排行榜失败，请稍后再试"

@views.route("/napcat/event", methods=["POST"])
@bypass_csrf_protection
def napcat_event():
    event_token = os.getenv("EVENT_TOKEN", "")

    query_token = request.args.get("token", "")
    auth_header = request.headers.get("Authorization", "")

    authorized = (
        query_token == event_token
        or auth_header == f"Bearer {event_token}"
    )

    if not event_token or not authorized:
        app.logger.warning("Rejected unauthorized NapCat event")
        return {"status": "unauthorized"}, 401

    data = request.get_json(silent=True) or {}

    app.logger.info("NapCat event: %s", data)
   
    # 1. 新人入群欢迎
    if (
        data.get("post_type") == "notice"
        and data.get("notice_type") == "group_increase"
    ):
        group_id = data.get("group_id")
        user_id = data.get("user_id")

        target_group = get_config("group_id")

        if (
            group_id
            and user_id
            and target_group
            and int(group_id) == int(target_group)
        ):
            send_napcat_group_message(
                group_id,
                [
                    {
                        "type": "at",
                        "data": {
                            "qq": str(user_id)
                        }
                    },
                    {
                        "type": "text",
                        "data": {
                            "text": (
                                " 欢迎加入工作室！🎉\n"
                                "📌 请先查看群公告与比赛规则\n"
                                "💬 有问题可以直接在群里交流"
                            )
                        }
                    }
                ]
            )

        return {"status": "ok"}

    # 2. 群消息命令
    if (
        data.get("post_type") == "message"
        and data.get("message_type") == "group"
    ):
        group_id = data.get("group_id")
        user_id = data.get("user_id")
        self_id = data.get("self_id")

        target_group = get_config("group_id")

        if not target_group:
            return {"status": "ignored"}

        if int(group_id or 0) != int(target_group):
            return {"status": "ignored"}

        message = data.get("message", [])

        # 判断消息中是否 @ 了机器人
        mentioned_bot = False
        text_parts = []

        if isinstance(message, list):
            for segment in message:
                segment_type = segment.get("type")
                segment_data = segment.get("data", {})

                if segment_type == "at":
                    qq = str(segment_data.get("qq", ""))

                    if qq == str(self_id):
                        mentioned_bot = True

                elif segment_type == "text":
                    text_parts.append(
                        str(segment_data.get("text", ""))
                    )

        # 合并纯文本
        command = "".join(text_parts).strip()

        app.logger.info(
            "QQ command: user=%s mentioned=%s command=%r",
            user_id,
            mentioned_bot,
            command
        )

        # 必须 @机器人 + #rank
        if mentioned_bot and command == "#rank":
            rank_text = build_rank_message()

            send_napcat_group_message(
                group_id,
                [
                    {
                        "type": "at",
                        "data": {
                            "qq": str(user_id)
                        }
                    },
                    {
                        "type": "text",
                        "data": {
                            "text": "\n" + rank_text
                        }
                    }
                ]
            )

    return {"status": "ok"}

这里注意一个事情，我这里的app其实current_app,因为我的顶部是

1
from flask import current_app as app

如果你没有引入或者没有as重命名，那就自己进行相应的更改

到这里所有的文件就已经更改完毕了，之后可以重新构建一下CTFd平台看看效果

参考

本文参考了以下文章

基于 Lagrange.Onebot 的 CTFd 解题播报机器人的搭建

[【CTFd 二次开发】解题播报机器人 V1 | Loora1N’s Blog | 鹭雨](https://loora1n.github.io/2022/10/11/【CTFd 二次开发】解题播报机器人 V1/)