package rdp import ( "context" "encoding/base64" "encoding/binary" "fmt" "io" "log" "strings" "sync" "time" "unicode/utf16" "ezops/pkg/rdp/channels/cliprdr" "ezops/pkg/rdp/channels/rdpdr" "ezops/pkg/rdp/display" "ezops/pkg/rdp/filestage" "ezops/pkg/rdp/trace" "ezops/pkg/rdp/transport" ) // channelSender 适配 cliprdr.Sender 接口:把虚拟通道 PDU 通过 MCS SendDataRequest // 发往指定 channelID,复用会话 writeMu 避免与 FastPath 输入并发写连接。// Logan.Li type channelSender struct { conn io.ReadWriter userID uint16 writeMu *sync.Mutex showProtocol bool } // 虚拟通道数据标志(MS-RDPBCGR 2.2.6.1 Virtual Channel PDU Header) const ( secFlagChannelPkt = 0x0004 channelFlagFirst = 0x00000001 channelFlagLast = 0x00000002 channelFlagShow = 0x00000010 channelChunkLength = 1600 // 单个 MCS PDU 的虚拟通道数据上限 ) // SendChannelPDU 按 MS-RDPBCGR 静态虚拟通道格式发送: // 所有静态虚拟通道发送都带 Channel PDU Header,showProtocol 只控制 flags。// Logan.Li func (s *channelSender) SendChannelPDU(channelID uint16, data []byte) error { s.writeMu.Lock() defer s.writeMu.Unlock() totalLen := uint32(len(data)) baseFlag := uint32(0) if s.showProtocol { baseFlag |= channelFlagShow } remaining := len(data) offset := 0 first := true for remaining > 0 { chunkSize := remaining if chunkSize > channelChunkLength { chunkSize = channelChunkLength } flag := baseFlag if first { flag |= channelFlagFirst first = false } remaining -= chunkSize if remaining == 0 { flag |= channelFlagLast } // 虚拟通道数据格式: [Channel PDU Header(totalLen+flags)] + [data]。// Logan.Li buf := make([]byte, 8+chunkSize) binary.LittleEndian.PutUint32(buf[0:], totalLen) // Channel PDU totalLen binary.LittleEndian.PutUint32(buf[4:], flag) // Channel PDU flags copy(buf[8:], data[offset:offset+chunkSize]) if err := transport.WriteX224Data(s.conn, buildMCSSendDataRequest(s.userID, channelID, buf)); err != nil { return err } offset += chunkSize } return nil } // SessionEngine 是自研完整 RDP 图形会话引擎 // Logan.Li type SessionEngine struct{} func (e *SessionEngine) Run(ctx context.Context, cfg Config, sink FrameSink, events <-chan ClientEvent) error { if cfg.Port == 0 { cfg.Port = 3389 } trace.Logf("engine=rdp connect target=%s:%d size=%dx%d security=%s", cfg.Host, cfg.Port, cfg.Width, cfg.Height, cfg.Security) _ = sink.SendFrame(ServerFrame{ Type: "status", Message: fmt.Sprintf("正在连接 RDP %s:%d", cfg.Host, cfg.Port), }) sess, err := OpenSession(ctx, cfg) if err != nil { trace.Logf("engine=rdp open failed target=%s:%d err=%v", cfg.Host, cfg.Port, err) _ = sink.SendFrame(ServerFrame{Type: "error", Message: "RDP 会话建立失败:" + err.Error()}) return err } defer func() { if c, ok := sess.Conn.(io.Closer); ok { _ = c.Close() } }() _ = sink.SendFrame(ServerFrame{Type: "status", Message: "RDP 图形会话已建立,等待桌面画面..."}) trace.Logf("engine=rdp open ok target=%s:%d", cfg.Host, cfg.Port) writeMu := &sync.Mutex{} input := NewInputSenderWithLock(sess.Conn, writeMu) receiver := NewFastPathReceiver() // 剪贴板虚拟通道:服务端授权了 cliprdr 通道才创建 Handler。// Logan.Li var clip *cliprdr.Handler var clipChannelID uint16 var drive *rdpdr.Handler var driveChannelID uint16 log.Printf("[RDP] mcs channels userID=%d global=%d granted=%d map=%v", sess.UserID, mcsGlobalChannelID, len(sess.ChannelNameByID), sess.ChannelNameByID) for id, name := range sess.ChannelNameByID { if name == cliprdr.ChannelName { clipChannelID = id } if name == rdpdr.ChannelName { driveChannelID = id } } if driveChannelID != 0 { drive = rdpdr.NewHandler("Go-client", "AIOps") drive.SetDownloadHandler(func(file rdpdr.Download) { if text, ok := decodeRDPClipboardFallback(file); ok { _ = sink.SendFrame(ServerFrame{Type: "clipboard", Text: text, Message: "已读取远端剪贴板"}) log.Printf("[RDP] rdpdr clipboard fallback ready len=%d", len(text)) return } token := filestage.Put(file.Name, file.Data, 30*time.Minute) _ = sink.SendFrame(ServerFrame{ Type: "rdp_file_download", Name: file.Name, URL: "/api/v1/rdp-staged/" + token, Size: int64(len(file.Data)), Message: "已捕获远端文件:" + file.Name, }) log.Printf("[RDP] rdpdr download ready name=%s size=%d token=%s", file.Name, len(file.Data), token) }) drive.Attach(&channelSender{conn: sess.Conn, userID: sess.UserID, writeMu: writeMu}, driveChannelID) log.Printf("[RDP] rdpdr channel attached id=%d", driveChannelID) drive.PreannounceDevices() go func() { time.Sleep(2 * time.Second) drive.PreannounceDevices() }() } if clipChannelID != 0 { clip = cliprdr.NewHandler(func(text string) { // 远端→本地:把远端剪贴板文本通过 WS clipboard 帧推给浏览器 _ = sink.SendFrame(ServerFrame{Type: "clipboard", Text: text}) log.Printf("[RDP] clipboard remote→local len=%d", len(text)) }) clip.Attach(&channelSender{conn: sess.Conn, userID: sess.UserID, writeMu: writeMu, showProtocol: true}, clipChannelID) log.Printf("[RDP] cliprdr channel attached id=%d", clipChannelID) go func() { time.Sleep(1500 * time.Millisecond) clip.Init() }() for _, ind := range sess.PendingChannelPDUs { if ind.channelID == clipChannelID { e.dispatchVirtualChannelPDU(ind, clip, clipChannelID) } if drive != nil && ind.channelID == driveChannelID { e.dispatchRdpdrPDU(ind, drive, driveChannelID) } } } else { log.Printf("[RDP] cliprdr channel NOT granted by server, clipboard sync disabled (channels=%v)", sess.ChannelNameByID) } // 会话激活后主动发起 cliprdr 握手,兼容等待客户端先声明能力的 Windows 版本。// Logan.Li readErr := make(chan error, 1) go func() { readErr <- e.readLoop(ctx, sess, receiver, sink, writeMu, clip, clipChannelID, drive, driveChannelID) }() go e.inputLoop(ctx, input, sink, events, clip, drive) for { select { case <-ctx.Done(): return ctx.Err() case err := <-readErr: if err != nil { _ = sink.SendFrame(ServerFrame{Type: "error", Message: err.Error()}) } return err } } } func (e *SessionEngine) inputLoop(ctx context.Context, input *InputSender, sink FrameSink, events <-chan ClientEvent, clip *cliprdr.Handler, drive *rdpdr.Handler) { for { select { case <-ctx.Done(): return case event, ok := <-events: if !ok { return } if event.Type != "mousemove" && event.Type != "mouse_move" { trace.Logf("engine=rdp input type=%s x=%d y=%d button=%d code=%s down=%t", event.Type, event.X, event.Y, event.Button, event.Code, event.Down) } if err := e.handleEvent(input, sink, event, clip, drive); err != nil { _ = sink.SendFrame(ServerFrame{Type: "status", Message: "输入事件失败:" + err.Error()}) } } } } func (e *SessionEngine) readLoop(ctx context.Context, sess *Session, receiver *FastPathReceiver, sink FrameSink, writeMu *sync.Mutex, clip *cliprdr.Handler, clipChannelID uint16, drive *rdpdr.Handler, driveChannelID uint16) error { framebuffer := display.NewFramebuffer(sess.Width, sess.Height) ticker := time.NewTicker(16 * time.Millisecond) defer ticker.Stop() refreshRequested := false usefulFrameSeen := false flushFrame := func() { frame, ok := framebuffer.FlushRGBA() if !ok { return } _ = sink.SendFrame(ServerFrame{ Type: "bitmap", X: frame.X, Y: frame.Y, Width: frame.Width, Height: frame.Height, RGBA: frame.RGBA, }) } sendBitmaps := func(rects []BitmapRect) { for _, rect := range rects { if rect.Width <= 0 || rect.Height <= 0 { continue } if isBlackFullScreen(rect, sess.Width, sess.Height) { if !refreshRequested { refreshRequested = true go func() { time.Sleep(400 * time.Millisecond) if err := e.requestRefresh(sess, writeMu); err != nil { trace.Logf("engine=rdp refresh request failed err=%v", err) } }() } if !usefulFrameSeen { // 登录初期的纯黑全屏帧通常是 RDP 激活噪声,跳过可避免浏览器闪黑。 // Logan.Li trace.Logf("engine=rdp skip startup black fullscreen") continue } } usefulFrameSeen = true framebuffer.Draw(display.Bitmap{ X: rect.DestLeft, Y: rect.DestTop, Width: rect.Width, Height: rect.Height, BitsPerPixel: rect.BitsPerPixel, Data: rect.Data, }) } } type packetResult struct { packet WirePacket err error } packetCh := make(chan packetResult, 1) go func() { for { pkt, err := ReadWirePacket(sess.Conn) select { case packetCh <- packetResult{packet: pkt, err: err}: case <-ctx.Done(): return } if err != nil { return } } }() for { select { case <-ctx.Done(): flushFrame() return ctx.Err() case <-ticker.C: flushFrame() case got := <-packetCh: if got.err != nil { flushFrame() return got.err } var rects []BitmapRect var err error if got.packet.FastPath { rects, err = receiver.Handle(got.packet.SecFlag, got.packet.Payload) } else { // SlowPath:先解析 channelID 判断是图形(global 1003)还是虚拟通道(cliprdr 等) // Logan.Li if ind, perr := parseMCSDataIndication(got.packet.Payload); perr == nil { if clip != nil && ind.channelID == clipChannelID { e.dispatchVirtualChannelPDU(ind, clip, clipChannelID) continue } if drive != nil && ind.channelID == driveChannelID { e.dispatchRdpdrPDU(ind, drive, driveChannelID) continue } if ind.channelID != mcsGlobalChannelID { log.Printf("[RDP] virtual channel ignored channel=%d name=%s payload=%d", ind.channelID, sess.ChannelNameByID[ind.channelID], len(ind.payload)) continue } logSlowPathSummary(ind, clipChannelID) } rects, err = receiver.HandleSlowPath(got.packet.Payload) } if err != nil { continue } if err := e.ackSurfaceFrames(sess, receiver.TakeFrameIDs(), writeMu); err != nil { trace.Logf("engine=rdp frame ack failed err=%v", err) } if len(rects) > 0 { trace.Logf("engine=rdp bitmap rects=%d fastpath=%t", len(rects), got.packet.FastPath) sendBitmaps(rects) } } } } func isBlackFullScreen(rect BitmapRect, width, height int) bool { if rect.DestLeft != 0 || rect.DestTop != 0 || rect.Width < width || rect.Height < height { return false } limit := len(rect.Data) if limit > 256*1024 { limit = 256 * 1024 } for i := 0; i < limit; i++ { if rect.Data[i] > 4 { return false } } return limit > 0 } func (e *SessionEngine) dispatchRdpdrPDU(ind mcsDataIndication, drive *rdpdr.Handler, driveChannelID uint16) { if drive == nil || ind.channelID != driveChannelID { return } prefixLen := len(ind.payload) if prefixLen > 12 { prefixLen = 12 } log.Printf("[RDP] rdpdr payload prefix=%x len=%d", ind.payload[:prefixLen], len(ind.payload)) if len(ind.payload) >= 4 && binary.LittleEndian.Uint16(ind.payload[0:2]) == 0x4472 { log.Printf("[RDP] rdpdr recv channel=%d rawLen=%d raw=true", ind.channelID, len(ind.payload)) drive.Process(ind.payload) return } if len(ind.payload) >= 8 && binary.LittleEndian.Uint32(ind.payload[0:4]) <= uint32(len(ind.payload)) && binary.LittleEndian.Uint16(ind.payload[4:6]) == 0x4472 { log.Printf("[RDP] rdpdr recv channel=%d rawLen=%d lengthOnlyHeader=true", ind.channelID, len(ind.payload)) drive.Process(ind.payload[4:]) return } if len(ind.payload) >= 8 && binary.LittleEndian.Uint16(ind.payload[4:6]) == 0x4472 { log.Printf("[RDP] rdpdr recv channel=%d rawLen=%d securityHeader=true", ind.channelID, len(ind.payload)) drive.Process(ind.payload[4:]) return } if p, flag, ok := parseVirtualChannelPDU(ind.payload); ok { log.Printf("[RDP] rdpdr recv channel=%d rawLen=%d flag=0x%x", ind.channelID, len(ind.payload), flag) if flag&channelFlagFirst != 0 && flag&channelFlagLast != 0 { drive.Process(p) } else { log.Printf("[RDP] rdpdr fragmented chunk flag=0x%x len=%d", flag, len(p)) } } } func (e *SessionEngine) dispatchVirtualChannelPDU(ind mcsDataIndication, clip *cliprdr.Handler, clipChannelID uint16) { if clip == nil || ind.channelID != clipChannelID { return } prefixLen := len(ind.payload) if prefixLen > 16 { prefixLen = 16 } log.Printf("[RDP] cliprdr payload prefix=%x len=%d", ind.payload[:prefixLen], len(ind.payload)) if p, flag, ok := parseVirtualChannelPDU(ind.payload); ok { log.Printf("[RDP] cliprdr recv channel=%d rawLen=%d flag=0x%x", ind.channelID, len(ind.payload), flag) if flag&channelFlagFirst != 0 && flag&channelFlagLast != 0 { clip.Process(p) } else { log.Printf("[RDP] cliprdr fragmented chunk flag=0x%x len=%d", flag, len(p)) } } else { log.Printf("[RDP] cliprdr parse failed channel=%d rawLen=%d", ind.channelID, len(ind.payload)) } } func parseVirtualChannelPDU(payload []byte) ([]byte, uint32, bool) { if len(payload) < 8 { return nil, 0, false } offset := 0 if hasVirtualChannelSecurityHeader(payload) { offset = 4 } totalLen := int(binary.LittleEndian.Uint32(payload[offset:])) flag := binary.LittleEndian.Uint32(payload[offset+4:]) dataStart := offset + 8 if totalLen < 0 || dataStart+totalLen > len(payload) { return nil, 0, false } return payload[dataStart : dataStart+totalLen], flag, true } func hasVirtualChannelSecurityHeader(payload []byte) bool { if len(payload) < 12 || binary.LittleEndian.Uint16(payload[0:2]) != secFlagChannelPkt { return false } totalLen := int(binary.LittleEndian.Uint32(payload[4:8])) flag := binary.LittleEndian.Uint32(payload[8:12]) if totalLen < 0 || 12+totalLen > len(payload) { return false } // 只有后移 4 字节后能看到合法 Channel PDU flags,才认为存在 security header。// Logan.Li return flag&(channelFlagFirst|channelFlagLast) != 0 } func logSlowPathSummary(ind mcsDataIndication, clipChannelID uint16) { prefixLen := len(ind.payload) if prefixLen > 24 { prefixLen = 24 } prefix := fmt.Sprintf("%x", ind.payload[:prefixLen]) type1 := uint16(0) type2 := byte(0) if len(ind.payload) >= 15 { type1 = binary.LittleEndian.Uint16(ind.payload[6:8]) type2 = ind.payload[14] } log.Printf("[RDP] slowpath packet channel=%d payload=%d sec=0x%04x type1=0x%04x type2=0x%02x prefix=%s (clipID=%d)", ind.channelID, len(ind.payload), ind.securityFlags, type1, type2, prefix, clipChannelID) } // isChannelData 判断一个 SlowPath MCS 包是否发往指定虚拟通道 ID。// Logan.Li func isChannelData(payload []byte, channelID uint16) bool { ind, err := parseMCSDataIndication(payload) if err != nil { return false } return ind.channelID == channelID } // extractChannelPayload 从 SlowPath MCS 包中提取虚拟通道负载(已去掉 MCS/SEC 头)。// Logan.Li func extractChannelPayload(payload []byte, channelID uint16) []byte { ind, err := parseMCSDataIndication(payload) if err != nil || ind.channelID != channelID { return nil } // 虚拟通道数据:payload 前 4 字节是 securityHeader(encrypt/decrypt 标志), // 之后的 raw data 即 CLIPRDR PDU。NLA 模式下通道不额外加密,直接取 data 段。 // Logan.Li if len(ind.payload) < 4 { return nil } return ind.payload[4:] } func (e *SessionEngine) requestRefresh(sess *Session, writeMu *sync.Mutex) error { body := buildRefreshRectBody(Config{Width: sess.Width, Height: sess.Height}) pdu := buildDataPDU(sess.UserID, sess.SharedID, pduType2RefreshRect, body) trace.Logf("engine=rdp refresh rect %dx%d", sess.Width, sess.Height) writeMu.Lock() err := transport.WriteX224Data(sess.Conn, buildMCSSendDataRequest(sess.UserID, mcsGlobalChannelID, pdu)) writeMu.Unlock() return err } func (e *SessionEngine) ackSurfaceFrames(sess *Session, frameIDs []uint32, writeMu *sync.Mutex) error { for _, frameID := range frameIDs { body := []byte{byte(frameID), byte(frameID >> 8), byte(frameID >> 16), byte(frameID >> 24)} pdu := buildDataPDU(sess.UserID, sess.SharedID, pduType2FrameAcknowledge, body) trace.Logf("engine=rdp frame ack frameID=%d", frameID) writeMu.Lock() err := transport.WriteX224Data(sess.Conn, buildMCSSendDataRequest(sess.UserID, mcsGlobalChannelID, pdu)) writeMu.Unlock() if err != nil { return err } } return nil } func (e *SessionEngine) handleEvent(input *InputSender, sink FrameSink, event ClientEvent, clip *cliprdr.Handler, drive *rdpdr.Handler) error { switch event.Type { case "mousemove", "mouse_move": return input.MouseMove(event.X, event.Y) case "mousedown", "mouse_down": return input.MouseButton(event.X, event.Y, event.Button, true) case "mouseup", "mouse_up": return input.MouseButton(event.X, event.Y, event.Button, false) case "wheel": return input.MouseWheel(event.X, event.Y, event.DeltaY) case "key": if event.Meta { event.Ctrl = true } if event.Down && (event.Ctrl || event.Meta) && (event.Code == "KeyC" || event.Code == "KeyX") && clip != nil { go func() { time.Sleep(250 * time.Millisecond) clip.RequestRemoteText() }() } if event.Code == "MetaLeft" || event.Code == "MetaRight" { event.Code = "ControlLeft" event.Key = "Control" } // 可打印字符按 Unicode 下发,避免远端键盘布局/修饰键状态导致审计有记录但远端无输入。 // Logan.Li if event.Down && len([]rune(event.Key)) == 1 && !event.Ctrl && !event.Alt && !event.Meta { return input.TypeText(event.Key) } if !event.Down && len([]rune(event.Key)) == 1 && !event.Ctrl && !event.Alt && !event.Meta { return nil } sc, ok := ScanCodeFromJS(event.Code) if !ok { return nil } return input.Key(sc, event.Down) case "remote_copy": if err := sendCopyShortcut(input); err != nil { return err } if clip != nil { go func() { time.Sleep(250 * time.Millisecond) clip.RequestRemoteText() }() } go func() { time.Sleep(900 * time.Millisecond) log.Printf("[RDP] clipboard explicit fallback auto_exec start") if err := input.AutoExec(buildClipboardFallbackCommand()); err != nil { log.Printf("[RDP] clipboard explicit fallback auto_exec failed: %v", err) return } log.Printf("[RDP] clipboard explicit fallback auto_exec sent") }() return sink.SendFrame(ServerFrame{Type: "status", Message: "已发送 Ctrl+C 到远端,等待剪贴板回传"}) case "clipboard": // 本地→远端:优先走 cliprdr 协议级同步(通知远端本地剪贴板变更), // 远端粘贴时主动拉取;cliprdr 不可用时回退到 Unicode 键盘模拟。// Logan.Li if event.Text == "" { return nil } if clip != nil { clip.SetLocalText(event.Text) } return input.TypeText(event.Text) case "rdp_file_upload": if drive == nil || event.Name == "" || event.Text == "" { return nil } data, err := base64.StdEncoding.DecodeString(event.Text) if err != nil { return err } drive.AddUpload(event.Name, data) return sink.SendFrame(ServerFrame{Type: "status", Message: "已放入 AIOps\\Upload:" + event.Name}) case "auto_exec": if event.Text == "" { return nil } _ = sink.SendFrame(ServerFrame{Type: "status", Message: "正在远端执行:" + event.Text}) return input.AutoExec(event.Text) case "resize": _ = sink.SendFrame(ServerFrame{Type: "status", Message: "分辨率变更请求已记录,重连后生效"}) return nil default: return nil } } func sendCopyShortcut(input *InputSender) error { // 工具按钮专用复制快捷键,避免复用前端 key 序列导致修饰键状态漂移。// Logan.Li if err := input.Key(0x001D, true); err != nil { return err } if err := input.Key(0x002E, true); err != nil { return err } if err := input.Key(0x002E, false); err != nil { return err } return input.Key(0x001D, false) } func buildClipboardFallbackCommand() string { ps := `$ErrorActionPreference='SilentlyContinue';$v='';try{$v=Get-Clipboard -Raw -ErrorAction Stop}catch{};if([string]::IsNullOrEmpty($v)){try{Add-Type -AssemblyName System.Windows.Forms;$v=[Windows.Forms.Clipboard]::GetText()}catch{}};if(-not [string]::IsNullOrEmpty($v)){$utf8=New-Object System.Text.UTF8Encoding -ArgumentList $false;$paths=@('\\tsclient\AIOps\Download\aiops-clipboard.txt','\\Go-client\AIOps\Download\aiops-clipboard.txt');foreach($p in $paths){try{[System.IO.File]::WriteAllText($p,$v,$utf8);break}catch{}}}` encoded := encodePowerShellCommand(ps) return "powershell -NoProfile -STA -WindowStyle Hidden -EncodedCommand " + encoded } func encodePowerShellCommand(command string) string { u16 := utf16.Encode([]rune(command)) raw := make([]byte, len(u16)*2) for i, v := range u16 { binary.LittleEndian.PutUint16(raw[i*2:], v) } return base64.StdEncoding.EncodeToString(raw) } func decodeRDPClipboardFallback(file rdpdr.Download) (string, bool) { if file.Name != "aiops-clipboard.txt" || len(file.Data) == 0 { return "", false } text := string(file.Data) text = strings.TrimPrefix(text, "\ufeff") return text, text != "" } // ProbeSessionFrames 用于诊断:建立会话后读取若干秒位图帧数量 // Logan.Li func ProbeSessionFrames(ctx context.Context, cfg Config, seconds int) (int, error) { sess, err := OpenSession(ctx, cfg) if err != nil { return 0, err } defer func() { if c, ok := sess.Conn.(io.Closer); ok { _ = c.Close() } }() receiver := NewFastPathReceiver() deadline := time.Now().Add(time.Duration(seconds) * time.Second) count := 0 for time.Now().Before(deadline) { if c, ok := sess.Conn.(interface{ SetReadDeadline(time.Time) error }); ok { _ = c.SetReadDeadline(time.Now().Add(500 * time.Millisecond)) } pkt, err := ReadWirePacket(sess.Conn) if err != nil { if time.Now().Before(deadline) { continue } return count, nil } var rects []BitmapRect if pkt.FastPath { rects, err = receiver.Handle(pkt.SecFlag, pkt.Payload) } else { rects, err = receiver.HandleSlowPath(pkt.Payload) } if err == nil { count += len(rects) } } if c, ok := sess.Conn.(interface{ SetReadDeadline(time.Time) error }); ok { _ = c.SetReadDeadline(time.Time{}) } return count, nil }