⚠️重要声明：此Demo仅用于学习网络编程，仅允许在自己授权的本机测试。未经授权部署到他人设备属于违法行为。 架构：中转服务端 <--> 被控客户端(Agent) <--> 监控GUI客户端 通信：TCP + JSON + \n做消息分隔；简单Token鉴权； 功能：agent上线上报IP；监控端看到在线列表；监控下发截图指令；agent截图返回图片base64；监控GUI显示截图。 运行顺序：先启动server.py，再启动agent.py（可多开模拟多台机器），最后启动controller_gui.py。 依赖：pip install pillow

业务目标：

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()

运行测试步骤（本机测试）

1. 安装依赖

pip install pillow

1. 启动中转服务端

python server.py

1. 新开终端，启动被控agent（可多开，模拟多台机器上线）

python agent.py

1. 新开终端，启动监控GUI

python controller_gui.py

1. 在GUI左侧看到agent上线；选中，点【截图选中主机】，右侧显示被控端屏幕截图。

部署到云服务器修改点

1. 修改agent.py和controller_gui.py中的SERVER_HOST为你的云服务器公网IP。
2. 云服务器安全组开放入站端口 8899(TCP)。

当前Demo局限（后续需要完善）

1. 报文明文传输，生产需要AES加密json内容；
2. 没有大图片分片；
3. 没有文件上传下载；
4. 没有鼠标键盘远程控制；
5. agent没有最小化后台静默，仅学习演示；
6. 断线重连逻辑简陋。

⚠️再次提醒：仅用于自己设备学习实验，未经许可部署他人设备触犯法律。