package controller import ( "bytes" "context" "encoding/base64" "encoding/binary" "errors" "image" "image/draw" "image/png" "log" "net/http" "strconv" "strings" "sync" "time" "ezops/middleware" "ezops/model" "ezops/pkg/crypto" rdpengine "ezops/pkg/rdp" "ezops/pkg/wshub" "github.com/gin-gonic/gin" "github.com/gorilla/websocket" "gorm.io/gorm" ) // RDPHandler 提供后端渲染型 RDP 会话,浏览器只负责 Canvas 与输入事件 // Logan.Li type RDPHandler struct { db *gorm.DB crypt *crypto.SecretBox wsMgr *wshub.WSManager engine rdpengine.Engine clients sync.Map // sessionID -> context.CancelFunc // Logan.Li live sync.Map // sessionID -> *rdpLiveBroadcaster // Logan.Li } const rdpAuditMaxText = 500 type sessionSecurityPolicy struct { ClipboardEnabled bool FileUploadEnabled bool FileDownloadEnabled bool SessionRecordEnabled bool } func loadSessionSecurityPolicy(db *gorm.DB) sessionSecurityPolicy { policy := sessionSecurityPolicy{ ClipboardEnabled: true, FileUploadEnabled: true, FileDownloadEnabled: true, SessionRecordEnabled: true, } var settings model.AuditSettings if err := db.First(&settings).Error; err != nil { return policy } policy.ClipboardEnabled = settings.ClipboardEnabled policy.FileUploadEnabled = settings.FileUploadEnabled policy.FileDownloadEnabled = settings.FileDownloadEnabled policy.SessionRecordEnabled = settings.SessionRecordEnabled return policy } func rdpEventBlockedByPolicy(event rdpengine.ClientEvent, policy sessionSecurityPolicy) (bool, string) { switch event.Type { case "clipboard", "remote_copy": if !policy.ClipboardEnabled { return true, "会话策略已禁用剪贴板" } case "file_upload", "rdp_file_upload": if !policy.FileUploadEnabled { return true, "会话策略已禁用文件上传" } case "file_download": if !policy.FileDownloadEnabled { return true, "会话策略已禁用文件下载" } } return false, "" } const rdpMaxWidth = 1920 const rdpMaxHeight = 1200 func NewRDPHandler(db *gorm.DB, masterKey string, wsMgr *wshub.WSManager) *RDPHandler { return &RDPHandler{ db: db, crypt: crypto.NewSecretBox(masterKey), wsMgr: wsMgr, engine: rdpengine.NewEngine(), } } func (h *RDPHandler) KillSession(sessionID uint) bool { if val, ok := h.clients.Load(sessionID); ok { cancel := val.(context.CancelFunc) cancel() h.clients.Delete(sessionID) return true } return false } func (h *RDPHandler) ServeLive(c *gin.Context) { id, err := strconv.ParseUint(c.Param("sessionId"), 10, 64) if err != nil { c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "无效的 RDP 会话ID"}) return } var session model.RDPSession if err := h.db.First(&session, id).Error; err != nil { c.JSON(http.StatusNotFound, model.Response{Code: 404, Message: "RDP 会话不存在"}) return } if session.Status != "active" { c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "RDP 会话已结束,请使用历史回放"}) return } val, ok := h.live.Load(uint(id)) if !ok { c.JSON(http.StatusNotFound, model.Response{Code: 404, Message: "RDP 实时流不存在"}) return } conn, info, err := h.wsMgr.Upgrade(c, wshub.TypeAudit, middleware.GetUserID(c), middleware.GetUsername(c)) if err != nil { return } defer h.wsMgr.Unregister(info) hub := val.(*rdpLiveBroadcaster) ch := hub.subscribe() defer hub.unsubscribe(ch) _ = writeRDPFrame(h.wsMgr, conn, rdpengine.ServerFrame{ Type: "session", Message: strconv.FormatUint(uint64(session.ID), 10), Width: session.Width, Height: session.Height, }) for { select { case frame, ok := <-ch: if !ok { return } if err := writeRDPFrame(h.wsMgr, conn, frame); err != nil { return } case <-c.Request.Context().Done(): return } } } func (h *RDPHandler) Serve(c *gin.Context) { serverID := c.Query("id") if serverID == "" { c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "缺少服务器ID"}) return } width := clampPositiveInt(c.Query("width"), 1280, rdpMaxWidth) height := clampPositiveInt(c.Query("height"), 800, rdpMaxHeight) var server model.Server if err := h.db.First(&server, serverID).Error; err != nil { c.JSON(http.StatusNotFound, model.Response{Code: 404, Message: "服务器不存在"}) return } normalizeServerRemoteAccess(&server) if !isRDPServer(server) { c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "该资产未配置为 RDP/Windows"}) return } rdpCred, err := h.loadRDPCredential(server) if err != nil { c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: err.Error()}) return } userID := middleware.GetUserID(c) username := middleware.GetUsername(c) policy := loadSessionSecurityPolicy(h.db) conn, info, err := h.wsMgr.Upgrade(c, "rdp", userID, username) if err != nil { return } log.Printf("[RDP] ws connected session pending server=%s ip=%s user=%s size=%dx%d", server.Hostname, server.IP, username, width, height) defer h.wsMgr.Unregister(info) var userIDPtr *uint if userID > 0 { userIDPtr = &userID } port := server.RDPPort if port == 0 { port = 3389 } session := model.RDPSession{ UserID: userIDPtr, Username: username, ServerID: server.ID, ServerName: server.Hostname, ServerIP: server.IP, ServerPort: port, ClientIP: c.ClientIP(), Width: width, Height: height, Status: "active", StartTime: time.Now(), } if err := h.db.Create(&session).Error; err != nil { _ = conn.WriteJSON(rdpengine.ServerFrame{Type: "error", Message: "创建 RDP 会话失败"}) return } ctx, cancel := context.WithCancel(c.Request.Context()) h.clients.Store(session.ID, cancel) liveHub := newRDPLiveBroadcaster() h.live.Store(session.ID, liveHub) defer func() { cancel() h.clients.Delete(session.ID) h.live.Delete(session.ID) liveHub.close() }() events := make(chan rdpengine.ClientEvent, 128) sink := &wsFrameSink{conn: conn, mgr: h.wsMgr, recorder: rdpengine.NewMemoryRecorder(500), live: liveHub, sessionID: session.ID, recordW: width, recordH: height} _ = sink.SendFrame(rdpengine.ServerFrame{Type: "session", Message: strconv.Itoa(int(session.ID)), Width: width, Height: height}) engineDone := make(chan error, 1) go func() { log.Printf("[RDP] engine start session=%d server=%s:%d size=%dx%d", session.ID, server.IP, port, width, height) engineDone <- h.engine.Run(ctx, rdpengine.Config{ Host: server.IP, Port: port, Username: rdpCred.Username, Password: rdpCred.Password, Domain: rdpCred.Domain, Security: rdpCred.Security, Width: width, Height: height, }, sink, events) }() status := "closed" errMsg := "" defer func() { cancel() end := time.Now() duration := int(end.Sub(session.StartTime).Seconds()) recordData := "" if policy.SessionRecordEnabled { recordData = sink.recorder.JSON() } updates := map[string]interface{}{ "status": status, "duration": duration, "end_time": end, "record_data": recordData, "record_size": len(recordData), } if errMsg != "" { updates["error"] = errMsg } if err := h.db.Model(&model.RDPSession{}).Where("id = ?", session.ID).Updates(updates).Error; err != nil { log.Printf("[RDP] 更新会话结束状态失败 (session_id=%d, status=%s): %v", session.ID, status, err) } }() audit := newRDPOperationRecorder(h.db, session, userIDPtr, username) defer audit.Flush() readDone := make(chan struct{}) modifiers := &rdpModifierState{} go func() { defer close(readDone) for { var event rdpengine.ClientEvent if err := conn.ReadJSON(&event); err != nil { return } if blocked, message := rdpEventBlockedByPolicy(event, policy); blocked { _ = sink.SendFrame(rdpengine.ServerFrame{Type: "status", Message: message}) continue } modifiers.Apply(&event) audit.Record(event) enqueueRDPEvent(events, event) } }() select { case err := <-engineDone: if err != nil && !errors.Is(err, context.Canceled) { status = "failed" errMsg = err.Error() log.Printf("[RDP] engine failed session=%d err=%v", session.ID, err) } else { log.Printf("[RDP] engine closed session=%d err=%v", session.ID, err) } case <-readDone: log.Printf("[RDP] ws read closed session=%d", session.ID) } } func enqueueRDPEvent(events chan rdpengine.ClientEvent, event rdpengine.ClientEvent) { select { case events <- event: return default: } if event.Type == "mousemove" || event.Type == "mouse_move" { // 队列已满时丢弃当前移动事件,避免旧坐标堆积拖慢点击/拖拽响应。 // Logan.Li return } // 关键输入不能被旧 mousemove 堵住,优先清理一批移动事件后再投递。 // Logan.Li for i := 0; i < cap(events)/2; i++ { select { case queued := <-events: if queued.Type != "mousemove" && queued.Type != "mouse_move" { select { case events <- queued: default: } } default: i = cap(events) } select { case events <- event: return default: } } select { case events <- event: case <-time.After(200 * time.Millisecond): log.Printf("[RDP] input queue congested, drop critical event type=%s", event.Type) } } type rdpModifierState struct { ctrl bool alt bool shift bool meta bool } // Apply 在服务端补齐修饰键状态,兼容旧前端未传 ctrlKey/altKey 的 RDP 按键事件。 // Logan.Li func (s *rdpModifierState) Apply(event *rdpengine.ClientEvent) { if event.Type != "key" { return } event.Ctrl = event.Ctrl || s.ctrl event.Alt = event.Alt || s.alt event.Shift = event.Shift || s.shift event.Meta = event.Meta || s.meta if event.Meta { event.Ctrl = true event.Meta = false } switch event.Code { case "ControlLeft", "ControlRight": s.ctrl = event.Down case "AltLeft", "AltRight": s.alt = event.Down case "ShiftLeft", "ShiftRight": s.shift = event.Down case "MetaLeft", "MetaRight": s.meta = event.Down } } type rdpOperationRecorder struct { db *gorm.DB session model.RDPSession userID *uint username string inputBuffer string lastInput time.Time } func newRDPOperationRecorder(db *gorm.DB, session model.RDPSession, userID *uint, username string) *rdpOperationRecorder { return &rdpOperationRecorder{db: db, session: session, userID: userID, username: username} } func (r *rdpOperationRecorder) Record(event rdpengine.ClientEvent) { now := time.Now() if r.inputBuffer != "" && now.Sub(r.lastInput) > 2*time.Second { r.Flush() } switch event.Type { case "key": r.recordKey(event, now) case "mousemove", "mouse_move", "mousedown", "mouse_down", "mouseup", "mouse_up", "wheel": return // 鼠标轨迹只进图形录屏,不写命令审计,避免 RDP 命令表噪声。 // Logan.Li case "clipboard": r.Flush() r.create("[RDP粘贴] " + truncateAuditText(event.Text, rdpAuditMaxText)) case "remote_copy": r.Flush() r.create("[RDP复制] Ctrl+C") case "file_upload": r.Flush() r.create("[RDP上传] " + truncateAuditText(event.Text, rdpAuditMaxText)) case "rdp_file_upload": r.Flush() r.create("[RDP上传到虚拟盘] " + truncateAuditText(event.Name, rdpAuditMaxText)) case "file_download": r.Flush() r.create("[RDP下载] " + truncateAuditText(event.Text, rdpAuditMaxText)) } } func (r *rdpOperationRecorder) recordKey(event rdpengine.ClientEvent, now time.Time) { if !event.Down { return } if event.Key == "Control" || event.Key == "Shift" || event.Key == "Alt" || event.Key == "Meta" { return } if event.Ctrl || event.Alt || event.Meta { r.Flush() if label, ok := rdpShortcutAuditLabel(event); ok { r.create(label) return } r.create("[RDP按键] " + rdpModifierLabel(event) + strings.ToUpper(event.Key)) return } if event.Code == "Enter" || event.Code == "NumpadEnter" { if strings.TrimSpace(r.inputBuffer) != "" { r.Flush() return } r.create("[RDP按键] Enter") return } if event.Code == "Backspace" { if len(r.inputBuffer) > 0 { r.inputBuffer = r.inputBuffer[:len(r.inputBuffer)-1] r.lastInput = now } return } if len(event.Key) == 1 { r.inputBuffer += event.Key r.lastInput = now if len(r.inputBuffer) >= rdpAuditMaxText { r.Flush() } return } if isAuditableRDPKey(event.Key) { r.Flush() r.create("[RDP按键] " + event.Key) } } // isAuditableRDPKey 只保留对 Windows 操作有审计意义的非文本按键。 // Logan.Li func isAuditableRDPKey(key string) bool { switch key { case "Tab", "Escape", "Delete", "Home", "End", "PageUp", "PageDown": return true } if strings.HasPrefix(key, "F") { n, err := strconv.Atoi(strings.TrimPrefix(key, "F")) return err == nil && n >= 1 && n <= 24 } return false } func rdpModifierLabel(event rdpengine.ClientEvent) string { parts := make([]string, 0, 3) if event.Ctrl { parts = append(parts, "Ctrl") } if event.Alt { parts = append(parts, "Alt") } if event.Meta { parts = append(parts, "Meta") } if len(parts) == 0 { return "" } return strings.Join(parts, "+") + "+" } func rdpShortcutAuditLabel(event rdpengine.ClientEvent) (string, bool) { if !event.Ctrl || event.Alt || event.Meta { return "", false } key := strings.ToUpper(event.Key) switch key { case "A": return "[RDP全选] Ctrl+A", true case "C": return "[RDP复制] Ctrl+C", true case "V": return "[RDP粘贴快捷键] Ctrl+V", true case "X": return "[RDP剪切] Ctrl+X", true default: return "", false } } func (r *rdpOperationRecorder) Flush() { if strings.TrimSpace(r.inputBuffer) == "" { r.inputBuffer = "" return } r.create("[RDP输入] " + truncateAuditText(r.inputBuffer, rdpAuditMaxText)) r.inputBuffer = "" } func (r *rdpOperationRecorder) create(command string) { cmd := model.SessionCommand{ SessionID: r.session.ID, SessionType: "rdp", UserID: r.userID, Username: r.username, ServerID: r.session.ServerID, ServerName: r.session.ServerName, Command: command, Timestamp: time.Now(), } if err := r.db.Create(&cmd).Error; err != nil { log.Printf("[RDPAudit] 记录 RDP 操作命令失败 (session_id=%d, user=%s, cmd=%q): %v", r.session.ID, r.username, command, err) } } func (h *RDPHandler) loadRDPCredential(server model.Server) (model.RDPSecretContent, error) { return loadRDPCredential(h.crypt, server) } func parsePositiveInt(raw string, fallback int) int { v, err := strconv.Atoi(raw) if err != nil || v <= 0 { return fallback } return v } func clampPositiveInt(raw string, fallback int, max int) int { v := parsePositiveInt(raw, fallback) if max > 0 && v > max { return max } return v } type wsFrameSink struct { conn *websocket.Conn mgr *wshub.WSManager mu sync.Mutex recorder *rdpengine.MemoryRecorder live *rdpLiveBroadcaster sessionID uint bitmaps int lastLog time.Time lastRecord time.Time recordFB *image.RGBA recordW int recordH int seenScreen bool } func (s *wsFrameSink) SendFrame(frame rdpengine.ServerFrame) error { if frame.Type != "bitmap" { s.recorder.Append(frame) } if s.live != nil { s.live.publish(frame) } if snap, ok := s.snapshotFrame(frame); ok { s.recorder.AppendSnapshot(snap) } if frame.Type == "bitmap" { s.bitmaps++ if s.bitmaps <= 3 || time.Since(s.lastLog) > 5*time.Second { log.Printf("[RDP] bitmap session=%d count=%d rect=%dx%d@%d,%d image=%d rgba=%d", s.sessionID, s.bitmaps, frame.Width, frame.Height, frame.X, frame.Y, len(frame.Image), len(frame.RGBA)) s.lastLog = time.Now() } } else if frame.Type == "status" || frame.Type == "error" || frame.Type == "session" { log.Printf("[RDP] frame session=%d type=%s msg=%q size=%dx%d", s.sessionID, frame.Type, frame.Message, frame.Width, frame.Height) } s.mu.Lock() defer s.mu.Unlock() return writeRDPFrame(s.mgr, s.conn, frame) } func (s *wsFrameSink) snapshotFrame(frame rdpengine.ServerFrame) (rdpengine.ServerFrame, bool) { if frame.Type != "bitmap" || len(frame.RGBA) == 0 || frame.Width <= 0 || frame.Height <= 0 { return rdpengine.ServerFrame{}, false } s.mu.Lock() defer s.mu.Unlock() if s.recordW <= 0 || s.recordH <= 0 { s.recordW = frame.Width + frame.X s.recordH = frame.Height + frame.Y if s.recordW < 1 || s.recordH < 1 { return rdpengine.ServerFrame{}, false } } if s.recordFB == nil { s.recordFB = image.NewRGBA(image.Rect(0, 0, s.recordW, s.recordH)) } rect := image.Rect(frame.X, frame.Y, frame.X+frame.Width, frame.Y+frame.Height).Intersect(s.recordFB.Bounds()) if rect.Empty() { return rdpengine.ServerFrame{}, false } src := image.NewRGBA(image.Rect(0, 0, frame.Width, frame.Height)) copy(src.Pix, frame.RGBA) draw.Draw(s.recordFB, rect, src, image.Point{X: rect.Min.X - frame.X, Y: rect.Min.Y - frame.Y}, draw.Src) if !s.seenScreen { if isMostlyBlackRGBA(s.recordFB, 0.985) { return rdpengine.ServerFrame{}, false } s.seenScreen = true } now := time.Now() if !s.lastRecord.IsZero() && now.Sub(s.lastRecord) < 500*time.Millisecond { return rdpengine.ServerFrame{}, false } s.lastRecord = now var buf bytes.Buffer if err := png.Encode(&buf, s.recordFB); err != nil { return rdpengine.ServerFrame{}, false } return rdpengine.ServerFrame{ Type: "bitmap", X: 0, Y: 0, Width: s.recordW, Height: s.recordH, Image: "data:image/png;base64," + base64.StdEncoding.EncodeToString(buf.Bytes()), }, true } // isMostlyBlackRGBA 判断录屏缓存是否仍是启动期黑屏,避免历史回放堆满无效黑帧。 // Logan.Li func isMostlyBlackRGBA(img *image.RGBA, threshold float64) bool { if img == nil || len(img.Pix) == 0 { return true } total := 0 black := 0 for i := 0; i+3 < len(img.Pix); i += 4 { total++ if img.Pix[i] <= 4 && img.Pix[i+1] <= 4 && img.Pix[i+2] <= 4 { black++ } } return total == 0 || float64(black)/float64(total) >= threshold } func writeRDPFrame(mgr *wshub.WSManager, conn *websocket.Conn, frame rdpengine.ServerFrame) error { mgr.SetWriteDeadline(conn) if frame.Type == "bitmap" && len(frame.RGBA) > 0 { return conn.WriteMessage(websocket.BinaryMessage, encodeRDPBitmapBinary(frame)) } return conn.WriteJSON(frame) } func encodeRDPBitmapBinary(frame rdpengine.ServerFrame) []byte { headerLen := 20 out := make([]byte, headerLen+len(frame.RGBA)) copy(out[0:4], []byte{'R', 'D', 'P', 'R'}) binary.LittleEndian.PutUint32(out[4:8], uint32(frame.X)) binary.LittleEndian.PutUint32(out[8:12], uint32(frame.Y)) binary.LittleEndian.PutUint32(out[12:16], uint32(frame.Width)) binary.LittleEndian.PutUint32(out[16:20], uint32(frame.Height)) copy(out[headerLen:], frame.RGBA) return out }