ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

Python+HTML 从零搭建 AWD 攻防平台:Flag 轮换与实时排行榜实战

Python+HTML 从零搭建 AWD 攻防平台:Flag 轮换与实时排行榜实战 简介这是一套基于 Python3 与 Django 开发的 AWD 网络攻防比赛裁判平台源码版本为 beta v2.0面向计算机相关专业的毕业设计、课程设计及项目开发学习者也适合想理解 CTF 攻防赛制与裁判系统实现原理的开发者参考。平台整体分为裁判机与靶机两部分通过特定接口完成靶机 flag 与服务器之间的通信涵盖赛题下发、flag 提交与校验等核心流程。资源包共 80 个文件以 18 个 py 源码、44 个 pyc 编译文件、6 个 py~ 备份、6 个 html 模板为主另含 sqlite3 数据库、README 说明文档、LICENSE 授权文件及 git 配置等压缩包约 95KB结构紧凑、便于快速部署与二次开发。目前已有 81 人学习关注。读者可据此掌握 Django 项目目录组织、接口通信设计与攻防平台业务逻辑并在此基础上扩展计分、队伍管理等模块是理解 AWD 赛制落地的实用参考。1. 从零搭一套 AWD 攻防平台为什么 Python HTML 是性价比最高的组合打 CTF 的人多少都碰过 AWDAttack With Defense赛制每支队伍守一台靶机既要修自己的漏洞又要打别人的服务Flag 每轮刷新分数实时滚动。真正让新手卡住的往往不是题目本身而是没有一套能跑起来的比赛环境——手动改 Flag、手动记分、手动判分一场训练赛下来裁判比选手还累。这个标题讲的就是用 Python 做后端、HTML 做前端自己搭一套 AWD 比赛平台把靶机管理、Flag 轮换、提交判分、实时排行榜这几件事自动化。它适合三类人想打 AWD 但找不到平台的入门选手、需要给课程设计交一套完整系统的学生、以及想理解攻防平台底层怎么运转的开发者。整套东西的核心不在界面多花哨而在于「Flag 怎么下发、怎么校验、怎么防作弊」这条链路能不能闭环。2. 平台架构拆解Python 后端与 HTML 前端各扛什么活2.1 为什么选 Flask 而不是 Django 做 AWD 后端AWD 平台的后端需求其实很集中接收 Flag 提交、校验对错、更新分数、给前端推排行榜。它不需要复杂的 ORM 关系也不需要后台管理系统用 Django 属于杀鸡用牛刀。Flask 的轻量在这里反而是优势——路由清晰、启动快、和 WebSocket 扩展Flask-SocketIO配合成熟几十行就能把核心接口跑起来。我一般会把后端拆成四层靶机管理层记录每队的 IP、端口、服务状态、Flag 管理层生成、下发、轮换、过期、判分引擎校验提交、计算得分、处理一血加成、通信层REST 接口 WebSocket 推送。这四层用 Flask 的 Blueprint 分开后期加功能不会互相污染。选 Flask 还有一个现实原因毕业设计和课程设计场景下代码要能被答辩老师看懂。Django 的 admin、middleware、settings 一堆约定反而增加解释成本。Flask 的app.py从头到尾读一遍就明白数据怎么流的。2.2 数据库表结构四张表撑起整个赛制AWD 的数据模型不复杂但字段设计错了后期很难改。下面是我实际用过的表结构用 SQLite 起步赛前切 MySQL 即可。-- 队伍表记录每支队伍的基础信息 CREATE TABLE teams ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, -- 队伍名排行榜展示用 token TEXT NOT NULL UNIQUE, -- 队伍登录凭证防冒名提交 score INTEGER DEFAULT 0, -- 当前总分 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- 靶机表一台靶机对应一支队伍的一个服务 CREATE TABLE targets ( id INTEGER PRIMARY KEY AUTOINCREMENT, team_id INTEGER NOT NULL, -- 归属队伍 host TEXT NOT NULL, -- 靶机 IP port INTEGER NOT NULL, -- 服务端口 service_name TEXT, -- 服务标识如 web1、pwn1 status TEXT DEFAULT running, -- running / down / patched FOREIGN KEY (team_id) REFERENCES teams(id) ); -- Flag 表每轮为每台靶机生成一个 Flag CREATE TABLE flags ( id INTEGER PRIMARY KEY AUTOINCREMENT, target_id INTEGER NOT NULL, value TEXT NOT NULL, -- Flag 字符串 round INTEGER NOT NULL, -- 属于第几轮 expired INTEGER DEFAULT 0, -- 是否已过期1 表示不可再提交 FOREIGN KEY (target_id) REFERENCES targets(id) ); -- 提交记录表所有提交都落库用于判重和审计 CREATE TABLE submissions ( id INTEGER PRIMARY KEY AUTOINCREMENT, team_id INTEGER NOT NULL, flag_value TEXT NOT NULL, is_correct INTEGER DEFAULT 0, -- 1 正确 0 错误 submitted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );teams.token这个字段容易被忽略但它是防作弊的关键提交 Flag 时必须带上本队 token否则别人拿到你的 Flag 可以替你交。flags.expired用来实现「只有当前轮的 Flag 算分」上一轮的 Flag 即使泄露也不再计分这是 AWD 赛制的核心规则。submissions表要建(team_id, flag_value)的联合索引判重时直接查避免同一 Flag 被反复提交刷分。2.3 前端页面清单五个 HTML 页面覆盖全流程前端不需要框架原生 HTML CSS JS 足够重点是页面职责要分清页面文件核心功能登录页login.html队伍 token 输入换取会话靶机面板dashboard.html展示本队靶机状态、当前 Flag提交页submit.html提交 Flag实时返回对错排行榜scoreboard.htmlWebSocket 实时刷新分数裁判后台admin.html开赛、轮换 Flag、查看提交日志scoreboard.html是唯一需要 WebSocket 的页面其余用普通 fetch 即可。很多教程一上来就全站 WebSocket结果调试时连接状态和业务状态混在一起排查起来很痛苦。我的做法是只有排行榜走长连接其他接口一律 REST出问题时能快速定位是网络层还是业务层。3. 核心功能实现Flag 轮换、提交判分与实时排行榜3.1 Flag 生成与轮换定时任务怎么写才不出错Flag 轮换是 AWD 平台最容易翻车的地方。常见做法是用APScheduler起一个后台调度器每 N 分钟执行一次轮换。轮换逻辑分三步把当前所有 Flag 标记为过期、为每台靶机生成新 Flag、把新 Flag 写入靶机通常通过 SSH 或预置的 agent 接口。from apscheduler.schedulers.background import BackgroundScheduler import uuid, time def rotate_flags(app): 每轮 Flag 轮换过期旧 Flag - 生成新 Flag - 下发到靶机 with app.app_context(): # 第一步把当前所有未过期的 Flag 标记为过期 db.execute(UPDATE flags SET expired 1 WHERE expired 0) # 第二步为每台运行中的靶机生成新 Flag targets db.execute( SELECT id FROM targets WHERE status running ).fetchall() current_round int(time.time() // 300) # 以 5 分钟为一轮 for t in targets: # Flag 格式AWD{随机串}前缀便于识别 flag_value AWD{ uuid.uuid4().hex[:16] } db.execute( INSERT INTO flags (target_id, value, round) VALUES (?, ?, ?), (t[id], flag_value, current_round) ) db.commit() # 第三步把新 Flag 推送到靶机伪代码按实际下发方式替换 push_flags_to_targets() # 启动调度器每 300 秒轮换一次 scheduler BackgroundScheduler() scheduler.add_job(funclambda: rotate_flags(app), triggerinterval, seconds300) scheduler.start()这里有几个参数必须想清楚。seconds300是轮换周期训练赛常用 3 到 5 分钟正式赛可能 10 分钟周期太短选手来不及利用漏洞太长则分数拉不开。uuid.uuid4().hex[:16]取 16 位十六进制碰撞概率可以忽略但如果你要更强的防猜测性可以换成secrets.token_hex(16)。current_round用时间戳整除周期来算好处是重启服务后轮次不会乱坏处是如果调度器卡顿可能出现两轮时间戳相同的情况——所以更稳妥的做法是单独维护一个round计数器存数据库。提示轮换任务一定要加异常捕获。如果push_flags_to_targets()中途失败前面的数据库更新已经提交会导致 Flag 和靶机实际状态不一致。建议把下发做成幂等操作失败后下一轮自动补发。3.2 提交判分接口三行校验挡住 90% 的作弊提交接口看起来简单但判分逻辑写错整场比赛的公信力就没了。核心校验只有三条Flag 是否存在且未过期、提交者是否为本队、该 Flag 是否已被本队提交过。from flask import request, jsonify app.route(/api/submit, methods[POST]) def submit_flag(): team_token request.json.get(token) flag_value request.json.get(flag, ).strip() # 校验 1队伍身份 team db.execute( SELECT id FROM teams WHERE token ?, (team_token,) ).fetchone() if not team: return jsonify({ok: False, msg: 无效队伍凭证}), 403 # 校验 2Flag 存在且未过期 flag_row db.execute( SELECT id, target_id FROM flags WHERE value ? AND expired 0, (flag_value,) ).fetchone() if not flag_row: return jsonify({ok: False, msg: Flag 无效或已过期}), 200 # 校验 3本队是否已提交过这个 Flag防重复刷分 dup db.execute( SELECT id FROM submissions WHERE team_id ? AND flag_value ? AND is_correct 1, (team[id], flag_value) ).fetchone() if dup: return jsonify({ok: False, msg: 该 Flag 已提交过}), 200 # 判分基础分 100一血额外加 50 base_score 100 first_blood db.execute( SELECT COUNT(*) AS c FROM submissions WHERE flag_value ? AND is_correct 1, (flag_value,) ).fetchone()[c] 0 gained base_score (50 if first_blood else 0) db.execute(UPDATE teams SET score score ? WHERE id ?, (gained, team[id])) db.execute( INSERT INTO submissions (team_id, flag_value, is_correct) VALUES (?, ?, 1), (team[id], flag_value) ) db.commit() return jsonify({ok: True, gained: gained, first_blood: first_blood})三个校验的顺序不能乱。先验身份再验 Flag避免未授权请求消耗数据库查询判重放在最后因为只有确认 Flag 有效才需要查提交记录。first_blood的判断用「该 Flag 是否已有正确提交」来算注意这里查的是所有队伍不是本队。基础分和一血加成的数值应该做成配置项不同比赛规则不一样硬编码在代码里后期改起来要命。注意返回错误时不要区分「Flag 不存在」和「Flag 已过期」统一返回模糊提示。否则攻击方可以通过错误信息判断自己拿到的 Flag 是过期了还是根本不存在变相获得情报。3.3 排行榜实时刷新WebSocket 推送的最小实现排行榜如果靠前端轮询几十支队伍同时在线时数据库压力不小。用 Flask-SocketIO 做服务端推送分数变化时主动广播前端只管渲染。from flask_socketio import SocketIO socketio SocketIO(app, cors_allowed_origins*) def broadcast_scoreboard(): 查询当前所有队伍分数并广播 rows db.execute( SELECT name, score FROM teams ORDER BY score DESC ).fetchall() data [{name: r[name], score: r[score]} for r in rows] socketio.emit(scoreboard_update, data) # 在 submit_flag 判分成功后调用 # broadcast_scoreboard()前端scoreboard.html里监听这个事件const socket io(http://localhost:5000); socket.on(scoreboard_update, (data) { const tbody document.querySelector(#rank-table tbody); tbody.innerHTML ; // 清空旧数据 data.forEach((team, idx) { const tr document.createElement(tr); tr.innerHTML td${idx 1}/tdtd${team.name}/tdtd${team.score}/td; tbody.appendChild(tr); }); });cors_allowed_origins*在开发阶段方便上线前必须改成具体域名否则任何页面都能连你的 WebSocket。广播时机也有讲究不要每次提交都广播高并发下会刷屏。我的做法是加一个 500ms 的节流或者只在分数实际变化时才触发广播。4. 避坑与排查AWD 平台上线前必须过的五道坎4.1 靶机 Flag 写入失败但数据库显示成功现象轮换日志显示新 Flag 已生成但选手在靶机上cat flag拿到的还是旧值。原因通常是下发环节用了异步 SSH数据库提交和实际写入之间存在时间差或者 SSH 连接超时被静默吞掉。解决把下发结果回写数据库给targets表加一个last_push_at字段轮换后检查该字段是否更新没更新就告警并重试。4.2 同一 Flag 被多队同时提交一血判断出错现象两支队伍几乎同时提交同一个 Flag结果两队都拿到了一血加成。原因是判分逻辑里「查询是否已有一血」和「插入提交记录」之间没有加锁并发下出现竞态。解决把这两步放进同一个数据库事务并对flags表加行锁MySQL 用SELECT ... FOR UPDATESQLite 则用BEGIN IMMEDIATE开启写事务。4.3 排行榜分数跳动刷新后名次对不上现象WebSocket 推来的分数和刷新页面后 REST 接口返回的分数不一致。原因是广播用的是内存里的临时数据而 REST 查的是数据库两者更新时机不同。解决广播前强制从数据库重新查询不要用提交接口里的局部变量拼数据。所有对外展示的分数只有一个数据源就是teams.score字段。4.4 选手提交接口被脚本高频调用数据库连接耗尽现象比赛开始后不久提交接口响应变慢日志里出现大量连接超时。原因是有人写了自动提交脚本每秒几十次请求。解决在提交接口加令牌桶限流每队每秒最多 5 次提交同时给submissions表的(team_id, submitted_at)建索引避免判重查询全表扫描。Flask 侧可以用Flask-Limiter快速加上限流。4.5 前端页面能打开但接口全部 403现象dashboard.html静态资源加载正常但所有/api/请求返回 403。原因通常是跨域配置或 token 传递方式不对——前端把 token 放在 body 里后端却从 header 取。解决统一约定 token 放在Authorization头里前端 fetch 时显式带上后端写一个before_request钩子统一校验不要在每个接口里重复写。5. 进阶技巧把平台从「能跑」推到「能扛比赛」平台跑通只是起点真正办一场比赛考验的是稳定性和可观测性。分享几个我踩过坑之后固定下来的习惯。第一给所有关键操作加审计日志。Flag 轮换、判分、分数变更这三类事件必须落库字段包括时间、操作类型、涉及队伍、结果。比赛结束后如果选手对分数有异议日志是唯一的后悔药。我一般单独建一张audit_log表用logging模块的 handler 直接写库不占用业务表。第二赛前做一次全链路压测。用locust或简单的 Python 脚本模拟 50 支队伍同时提交观察三个指标提交接口 P99 延迟、WebSocket 广播延迟、数据库连接数峰值。如果 P99 超过 500ms说明判分逻辑里有慢查询回去看submissions表的索引。压测脚本本身不用复杂循环调提交接口就行关键是量要够。第三Flag 下发通道做双保险。常见做法是 SSH 直连靶机写文件但比赛网络抖动时 SSH 会断。我的方案是靶机侧跑一个轻量 agent监听一个内部端口平台通过 HTTP 把 Flag 推给 agentagent 负责写文件并返回确认。这样平台不依赖 SSH靶机重启后 agent 自动重连Flag 状态也能主动上报。第四分数计算规则做成可配置。不同比赛对一血、连杀、服务存活时长的计分方式不同把这些规则抽到一个scoring.yaml里判分引擎读配置执行。下面是一个配置示例# scoring.yaml base_score: 100 # 每个有效 Flag 基础分 first_blood_bonus: 50 # 一血额外加分 service_alive_bonus: 10 # 每轮服务存活加分 max_submit_per_sec: 5 # 每队提交限流判分时读这个文件改规则不用动代码。答辩时老师问「如果规则变了怎么办」直接改 YAML 演示一遍比解释代码逻辑有说服力。最后说个验证方法平台搭好后自己开两个浏览器窗口一个当攻击方一个当防守方完整走一遍「拿 Flag → 提交 → 看排行榜变化 → 等下一轮 Flag 过期 → 再提交旧 Flag 应失败」的流程。这个手动回归测试能覆盖 80% 的逻辑错误比写单元测试快得多。我每次改完判分逻辑都会跑一遍已经成了肌肉记忆。希望帮到你。本文还有配套的精品资源点击获取
返回列表