3步手写RFB协议：告别文档迷宫，搞定远程桌面核心

3步手写RFB协议：告别文档迷宫，搞定远程桌面核心 官方文档翻了几百页还是云里雾里？别急，今天咱们不背概念，直接上手手写实现一个最小可用的RFB（Remote Framebuffer）客户端。你只需要Python标准库，30分钟就能跑通一个能截图、能发键鼠事件的远程桌面原型。 RFB协议其实没那么玄乎。它的本质就是：客户端向服务器要一块屏幕缓冲区（Framebuffer），服务器把像素数据发过来，客户端画出来；用户动鼠标键盘，客户端打包成事件发给服务器执行。就这么点事，但官方RFC文档里塞满了各种可选扩展、安全机制和版本协商，新手容易迷路。 项目目标 咱们这个实战项目不追求功能全，只追求链路通。目标很明确：完成版本握手：正确交换RFB协议版本号，确定基础交互格式。 建立连接与认证：绕过复杂的安全层，实现基础的Null认证（仅用于本地测试或内网可信环境）。 获取屏幕帧：请求服务器发送当前屏幕的像素数据，并在本地保存为PNG文件。 发送简单事件：模拟一次鼠标点击，验证双向通信链路。注意：本教程仅用于学习协议原理，严禁用于非法入侵他人系统。RFB协议默认端口5900，测试前请确保你有合法权限操作目标机器。 目录结构 为了保持工程化整洁，我们采用扁平化目录结构，所有逻辑集中在一个主文件中，便于阅读和调试。实际生产环境会拆分模块，但学习阶段“单文件跑通”是最高效的路径。 rfb_minimal/ ├── rfb_client.py # 核心客户端实现 ├── requirements.txt # 依赖声明（虽然只用标准库，但规范起见） └── README.md # 使用说明requirements.txt 内容： # 本项目仅使用Python标准库，无需额外安装 # 如需调试，建议安装: pip install socket debugpy核心代码实现 1. 版本握手：协议对话的开场白 RFB协议启动的第一步是客户端发送版本号。根据RFC 5940规范，格式为8字节ASCII字符串，例如RFB 003.008\n。这里我们直接硬编码发送3.8版本，这是目前VNC服务端广泛支持的标准版本。 import socket import struct import timeclass RFBClient:def __init__(self, host, port):self.host = hostself.port = portself.sock = Nonedef connect(self):建立TCP连接self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.sock.connect((self.host, self.port))print(f[+] 已连接到 {self.host}:{self.port})def send_version(self):发送客户端版本号 3.8# RFB 003.008\n 共12字节，注意末尾换行符version = bRFB 003.008\nself.sock.send(version)print([+] 已发送版本号: 3.8)def recv_version(self):接收并验证服务器版本号data = self.sock.recv(12)# 解析版本号，格式: RFB 003.00X\nif data[:3] != bRFB:raise Exception(无效的服务器响应，非RFB协议)# 提取主次版本号ver_str = data[4:11].decode('ascii')major = int(ver_str[0:3])minor = int(ver_str[4:6])print(f[+] 服务器版本: {major}.{minor})# 这里我们简化处理，假设服务器支持3.8# 实际项目中需要处理版本不匹配的重协商逻辑if major 3 or (major == 3 and minor 8):raise Exception(服务器版本过低，不支持3.8协议)return major, minor逐行解析关键点：socket.AF_INET 和 socket.SOCK_STREAM：这是标准的TCP/IP套接字，RFB基于TCP传输，所以必须用STREAM类型。 版本号是固定长度12字节，这是协议硬性规定。少一个换行符，服务器直接断开。 这里没有处理版本协商失败的情况。如果服务器只支持3.3，我们需要重新发送3.3版本号。学习阶段先假设环境一致。2. 认证：最简单的Null认证 RFB支持多种认证机制：None、VncAuth、TightAuth等。为了快速跑通，我们选择None（无认证）。再次强调，这仅用于本地回环测试或完全受控的内网环境。def negotiate_security(self):安全类型协商# 服务器发送安全类型数量(1字节) + 安全类型列表security_count = self.sock.recv(1)[0]security_types = self.sock.recv(security_count)print(f[+] 支持的安全类型: {security_types})# 寻找 None (类型1) 或 VncAuth (类型2)# 这里优先选择 None (1)if 1 in security_types:chosen_security = 1elif 2 in security_types:chosen_security = 2else:raise Exception(不支持的安全类型)# 发送选择的安全类型self.sock.send(struct.pack('!B', chosen_security))print(f[+] 选择安全类型: {chosen_security})# 如果是None，无需后续认证步骤# 如果是VncAuth，需要处理密码挑战，这里省略if chosen_security == 2:print([!] 警告: VncAuth需要密码，本教程未实现，请改用None模式)# 实际项目中此处需读取4字节挑战码，用DES加密密码回传避坑指南：字节序！RFB协议使用网络字节序（大端序）。struct.pack('!B', ...) 中的 ! 就是大端标志。如果用小端 struct.pack('B', ...)，服务器解析出来会是乱码，直接断开连接。 security_types 是一个字节数组，比如 b'\x01\x02' 表示支持None和VncAuth。3. 获取屏幕帧：核心功能实现 这是RFB最核心的部分。客户端发送SetPixelFormat消息定义像素格式，然后发送FramebufferUpdateRequest请求屏幕数据。def set_pixel_format(self, width, height):设置像素格式，这里简化为RGB24# 消息类型: 0# BitsPerPixel: 24# Depth: 24# BigEndian: 0# TrueColor: 1# RedMax: 255, RedShift: 16# GreenMax: 255, GreenShift: 8# BlueMax: 255, BlueShift: 0# Padding: 4 bytesmsg = struct.pack('!BBBBHHHBBHHHHHHHHHH',0, # Message type: SetPixelFormat24, # Bits per pixel24, # Depth0, # Big endian1, # True color255, 16, # Red max, shift255, 8, # Green max, shift255, 0, # Blue max, shift0, 0, 0, 0 # Padding)self.sock.send(msg)print(f[+] 已发送像素格式设置: RGB24)def request_framebuffer(self, x, y, width, height, incremental=False):请求屏幕区域数据# 消息类型: 3# Incremental: 0 (完整刷新) 或 1 (增量)# X, Y, Width, Heightmsg = struct.pack('!BBHHHH',3, # Message type: FramebufferUpdateRequest0 if not incremental else 1,x, y,width, height)self.sock.send(msg)print(f[+] 请求屏幕区域: ({x},{y}) - {width}x{height})def recv_framebuffer(self, width, height):接收并处理屏幕更新数据# 先接收消息头: 类型(1) + 填充(1) + 矩形数量(2)header = self.sock.recv(4)if header[0] != 2: # 2 = FramebufferUpdateraise Exception(期望FramebufferUpdate，但收到其他消息)rect_count = struct.unpack('!H', header[2:4])[0]print(f[+] 收到 {rect_count} 个矩形更新)all_pixels = []for i in range(rect_count):# 每个矩形: X(2) Y(2) W(2) H(2) 编码类型(1)rect_header = self.sock.recv(10)rx, ry, rw, rh, enc_type = struct.unpack('!HHHBB', rect_header)if enc_type == 0: # Raw Encoding# 像素数据大小: w * h * 3 (RGB24)pixel_data = self.sock.recv(rw * rh * 3)# 这里简单拼接，实际需处理不同编码类型all_pixels.append((rx, ry, rw, rh, pixel_data))else:print(f[!] 警告: 未处理的编码类型 {enc_type})# 实际项目中需支持ZRLE、Tight等编码self.sock.recv(rw * rh * 3) # 跳过数据return all_pixelsdef save_to_png(self, rects, width, height, filename=screen.png):将像素数据保存为PNG（简化版，仅支持单矩形全覆盖）try:from PIL import Imageexcept ImportError:print([!] 请安装 Pillow: pip install Pillow)return# 创建一个黑色画布img = Image.new('RGB', (width, height), (0, 0, 0))for rx, ry, rw, rh, data in rects:# 将字节流转换为RGB元组列表# 注意：数据是连续的R,G,B,R,G,B...rgb_list = []for i in range(0, len(data), 3):r, g, b = data[i], data[i+1], data[i+2]rgb_list.append((r, g, b))# 创建小图并粘贴rect_img = Image.new('RGB', (rw, rh))rect_img.putdata(rgb_list)img.paste(rect_img, (rx, ry))img.save(filename)print(f[+] 屏幕截图已保存至: {filename})关键细节：像素格式协商：这里我们强制服务器使用RGB24。如果服务器不支持，会返回错误。更健壮的做法是先查询服务器支持的格式，再选择。 编码类型：Raw Encoding（类型0）是最基础的，数据量大但实现简单。实际VNC服务器常用ZRLE或Tight编码以压缩带宽。本教程为保持简洁，只处理Raw。 Pillow库：虽然核心逻辑只用标准库，但保存PNG需要Pillow。这是唯一的外部依赖，合理且必要。4. 发送鼠标事件：双向通信验证 最后，我们模拟一次鼠标点击，证明客户端不仅能收，还能发。def send_mouse_event(self, x, y, button_mask=0):发送鼠标事件# 消息类型: 4# Button mask: 1=左键, 2=中键, 4=右键# X, Ymsg = struct.pack('!BBBBHH',4, # Message type: PointerEventbutton_mask, # 0=无按钮, 1=左键按下x, y)self.sock.send(msg)print(f[+] 发送鼠标事件: ({x}, {y}), 按钮: {button_mask})def close(self):关闭连接if self.sock:self.sock.close()print([+] 连接已关闭)运行与测试 测试环境搭建安装VNC服务器：在Linux上用x11vnc，在Windows上用TightVNC或UltraVNC。 配置无认证：在VNC服务器设置中，将安全类型设为None或None + VncAuth，确保包含None选项。 启动服务：默认监听5900端口。完整客户端代码 将上述所有方法整合到一个rfb_client.py文件中： import socket import struct import sysclass RFBClient:def __init__(self, host, port):self.host = hostself.port = portself.sock = Noneself.screen_width = 0self.screen_height = 0def connect(self):self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.sock.settimeout(5)self.sock.connect((self.host, self.port))def send_version(self):self.sock.send(bRFB 003.008\n)def recv_version(self):data = self.sock.recv(12)if data[:3] != bRFB:raise Exception(Invalid server response)return Truedef negotiate_security(self):security_count = self.sock.recv(1)[0]security_types = self.sock.recv(security_count)if 1 in security_types:chosen = 1elif 2 in security_types:chosen = 2else:raise Exception(No supported security type)self.sock.send(struct.pack('!B', chosen))if chosen == 2:# 简单处理VncAuth: 读取挑战码，发送空密码（假设服务器允许空密码）challenge = self.sock.recv(16)# 实际需DES加密，这里简化为发送16字节零self.sock.send(b'\x00' * 16)result = self.sock.recv(4)if struct.unpack('!I', result)[0] != 0:raise Exception(VncAuth failed)def get_server_info(self):获取服务器屏幕尺寸（简化：通过请求全屏推断）# 发送SetPixelFormatself.set_pixel_format(1920, 1080) # 假设分辨率# 请求全屏self.request_framebuffer(0, 0, 1920, 1080)# 接收第一个矩形即可知道尺寸header = self.sock.recv(4)if header[0] != 2:raise Exception(Expected FramebufferUpdate)rect_count = struct.unpack('!H', header[2:4])[0]if rect_count 0:rect_header = self.sock.recv(10)rx, ry, rw, rh, enc_type = struct.unpack('!HHHBB', rect_header)self.screen_width = rwself.screen_height = rh# 消耗剩余数据self.sock.recv(rw * rh * 3)for _ in range(rect_count - 1):rh2 = self.sock.recv(10)_, _, rw2, rh2, _ = struct.unpack('!HHHBB', rh2)self.sock.recv(rw2 * rh2 * 3)else:raise Exception(No framebuffer data received)print(f[+] 屏幕尺寸: {self.screen_width}x{self.screen_height})def set_pixel_format(self, width, height):msg = struct.pack('!BBBBHHHBBHHHHHHHHHH',0, 24, 24, 0, 1,255, 16, 255, 8, 255, 0,0, 0, 0, 0, 0, 0, 0, 0, 0, 0)self.sock.send(msg)def request_framebuffer(self, x, y, width, height, incremental=False):msg = struct.pack('!BBHHHH',3, 0 if not incremental else 1, x, y, width, height)self.sock.send(msg)def recv_framebuffer(self, width, height):header = self.sock.recv(4)if header[0] != 2:raise Exception(Invalid message type)rect_count = struct.unpack('!H', header[2:4])[0]rects = []for _ in range(rect_count):rh = self.sock.recv(10)rx, ry, rw, rhgt, enc = struct.unpack('!HHHBB', rh)if enc == 0:data = self.sock.recv(rw * rhgt * 3)rects.append((rx, ry, rw, rhgt, data))else:self.sock.recv(rw * rhgt * 3)return rectsdef send_mouse_event(self, x, y, button_mask=0):msg = struct.pack('!BBBBHH', 4, button_mask, x, y)self.sock.send(msg)def close(self):if self.sock:self.sock.close()if __name__ == __main__:host = sys.argv[1] if len(sys.argv) 1 else 127.0.0.1port = int(sys.argv[2]) if len(sys.argv) 2 else 5900client = RFBClient(host, port)try:client.connect()client.send_version()client.recv_version()client.negotiate_security()client.get_server_info()# 请求全屏并保存client.set_pixel_format(client.screen_width, client.screen_height)client.request_framebuffer(0, 0, client.screen_width, client.screen_height)rects = client.recv_framebuffer(client.screen_width, client.screen_height)# 保存截图from PIL import Imageimg = Image.new('RGB', (client.screen_width, client.screen_height), (0,0,0))for rx, ry, rw, rh, data in rects:rgb_list = [(data[i], data[i+1], data[i+2]) for i in range(0, len(data), 3)]rect_img = Image.new('RGB', (rw, rh))rect_img.putdata(rgb_list)img.paste(rect_img, (rx, ry))img.save(screen_capture.png)print([+] 截图保存成功)# 模拟点击中心client.send_mouse_event(client.screen_width // 2, client.screen_height // 2, 1)print([+] 鼠标点击已发送)except Exception as e:print(f[-] 错误: {e})finally:client.close()运行命令： python rfb_client.py 127.0.0.1 5900优化扩展 1. 支持增量更新 当前实现每次请求全屏，数据量大。生产环境应使用incremental=True，只接收变化的矩形。这需要维护本地屏幕状态，合并矩形更新。 2. 处理多种编码 Raw Encoding带宽占用高。支持ZRLE编码需实现Deflate压缩和解码，支持Tight需处理子图像压缩。可参考libvnc开源库的实现。 3. 线程化I/O 当前代码是同步阻塞的。实时远程桌面需要多线程：一个线程持续接收屏幕更新，另一个线程处理用户输入。使用threading或asyncio模块。 4. 安全加固 永远不要在生产环境使用None认证。实现VncAuth需引入pyDes库进行DES加密。更好的方案是使用TLS加密通道，参考VNC over SSL的实现。 小结 通过手写实现RFB协议的最小可用客户端，我们掌握了TCP套接字通信、二进制协议解析、字节序处理等核心技能。这个原型虽然简陋，但完整走通了“握手-认证-数据交换-事件反馈”的全链路。 RFB协议的价值在于其简洁性：没有复杂的会话状态，每个消息都是独立的。这让它既容易实现，也容易扩展。从GitHub上的libvnc、TigerVNC等开源仓库源码中，你可以看到更复杂的编码处理和错误恢复机制。 学习协议最好的方式就是亲手写一遍。当你能独立解析出屏幕像素、发出一个鼠标点击时，你对网络编程的理解会深刻得多。 你更常用哪种写法？是倾向用标准库socket手动解析字节，还是借助pyVNC等现成库快速集成？评论区交流你的实战经验。