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

官方文档翻了几百页还是云里雾里？别急，今天咱们不背概念，直接上手手写实现一个最小可用的RFB（Remote Framebuffer）客户端。你只需要Python标准库，30分钟就能跑通一个能截图、能发键鼠事件的远程桌面原型。

RFB协议其实没那么玄乎。它的本质就是：客户端向服务器要一块屏幕缓冲区（Framebuffer），服务器把像素数据发过来，客户端画出来；用户动鼠标键盘，客户端打包成事件发给服务器执行。就这么点事，但官方RFC文档里塞满了各种可选扩展、安全机制和版本协商，新手容易迷路。

项目目标

咱们这个实战项目不追求功能全，只追求链路通。目标很明确：

1. 完成版本握手：正确交换RFB协议版本号，确定基础交互格式。
2. 建立连接与认证：绕过复杂的安全层，实现基础的Null认证（仅用于本地测试或内网可信环境）。
3. 获取屏幕帧：请求服务器发送当前屏幕的像素数据，并在本地保存为PNG文件。
4. 发送简单事件：模拟一次鼠标点击，验证双向通信链路。

注意：本教程仅用于学习协议原理，严禁用于非法入侵他人系统。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 = b"RFB 003.008\n"self.sock.send(version)print("[+] 已发送版本号: 3.8")def recv_version(self):"""接收并验证服务器版本号"""data = self.sock.recv(12)# 解析版本号，格式: RFB 003.00X\nif data[:3] != b"RFB":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("[+] 连接已关闭")

运行与测试

测试环境搭建

1. 安装VNC服务器：在Linux上用x11vnc，在Windows上用TightVNC或UltraVNC。
2. 配置无认证：在VNC服务器设置中，将安全类型设为"None"或"None + VncAuth"，确保包含None选项。
3. 启动服务：默认监听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(b"RFB 003.008\n")def recv_version(self):data = self.sock.recv(12)if data[:3] != b"RFB":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.1"port = 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等现成库快速集成？评论区交流你的实战经验。