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

napcat部署

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

sudo docker pull mlikiowa/napcat-docker:latest

安装好之后运行

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

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

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

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

这里的sever token要保存下来

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

sudo docker network ls

然后将Napcat接入CTFd的网络

sudo docker network connect aegis_default napcat

检查是否接入

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

看到类似

bridge
aegis_default

就可以了

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

napcat:3000

同理 NapCat 可以使用：

aegis-ctfd-1:8000

访问 CTFd。

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

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

CTFd文件修改

general.html

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

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

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

加入：

<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生成检查脚本）(≧▽≦)

执行：

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

正常会返回：

{
    "status": "ok",
    "retcode": 0,
    "data": {
        ...
    }
}

这说明：

CTFd → NapCat Docker网络正常
NapCat HTTP Server正常
Token正常
QQ登录正常

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

challenges.py

位置在

/CTFd/api/v1/challenges.py

顶部引入

import os
import requests

from flask import current_app

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

user_name = user.name
challenge_name = challenge.name

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

qq_number = None

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

    if candidate.isdigit():
        qq_number = candidate

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

例如：

123456789@qq.com

会得到：

123456789

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

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

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}」！"
        )

最终完整发送逻辑：

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"
            )

这段代码的位置是在

clear_standings()

clear_challenges()

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

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

找到Patch函数

将它进行修改

    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

在头部确保

import os
import requests

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

            # 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)

效果大致如下

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然后机器人能返回排名,还有就是自动欢迎新成员

在头部引入

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

在文件末尾添加

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,因为我的顶部是

from flask import current_app as app

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

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

参考

本文参考了以下文章

基于 Lagrange.Onebot 的 CTFd 解题播报机器人的搭建
【CTFd 二次开发】解题播报机器人 V1 | Loora1N’s Blog | 鹭雨