feat(pam-rdp): record RDP sessions

packages/api/model.go

@@ -958,8 +958,7 @@ type UploadSessionLogEntry struct {
 	Output    string    `json:"output"`
 }
 
-// UploadTerminalEvent represents a terminal session event for upload
-type UploadTerminalEvent struct {
+type UploadSessionEvent struct {
 	Timestamp   time.Time `json:"timestamp"`
 	EventType   string    `json:"eventType"`
 	ChannelType string    `json:"channelType,omitempty"`
@@ -979,7 +978,7 @@ type UploadHttpEvent struct {
 }
 
 type UploadPAMSessionLogsRequest struct {
-	Logs interface{} `json:"logs"` // Can be []UploadSessionLogEntry or []UploadTerminalEvent
+	Logs interface{} `json:"logs"` // Can be []UploadSessionLogEntry or []UploadSessionEvent
 }
 
 type RelayHeartbeatRequest struct {

packages/gateway-v2/gateway.go

@@ -89,6 +89,7 @@ type GatewayConfig struct {
 type pamSessionEntry struct {
 	cancel       context.CancelFunc
 	conn         *tls.Conn
+	done         chan struct{} // closed when HandlePAMProxy has fully returned for this entry
 	lastActivity atomic.Int64
 }
 
@@ -170,7 +171,7 @@ func NewGateway(config *GatewayConfig) (*Gateway, error) {
 // RegisterPAMSession registers an active PAM proxy connection for cancellation support
 // Returns a function that handlers should call when data flows through the connection
 func (g *Gateway) RegisterPAMSession(sessionID string, cancel context.CancelFunc, conn *tls.Conn) func() {
-	entry := &pamSessionEntry{cancel: cancel, conn: conn}
+	entry := &pamSessionEntry{cancel: cancel, conn: conn, done: make(chan struct{})}
 	entry.lastActivity.Store(time.Now().Unix())
 
 	g.pamSessionsMu.Lock()
@@ -189,23 +190,55 @@ func (g *Gateway) RegisterPAMSession(sessionID string, cancel context.CancelFunc
 // The proxy is cleaned up on session cancellation or gateway shutdown.
 func (g *Gateway) DeregisterPAMSession(sessionID string, conn *tls.Conn) bool {
 	g.pamSessionsMu.Lock()
-	defer g.pamSessionsMu.Unlock()
-
 	entries, exists := g.pamSessions[sessionID]
 	if !exists {
+		g.pamSessionsMu.Unlock()
 		return false
 	}
+	var removed *pamSessionEntry
 	for i, e := range entries {
 		if e.conn == conn {
+			removed = e
 			g.pamSessions[sessionID] = append(entries[:i], entries[i+1:]...)
 			break
 		}
 	}
-	if len(g.pamSessions[sessionID]) == 0 {
+	isLast := len(g.pamSessions[sessionID]) == 0
+	if isLast {
 		delete(g.pamSessions, sessionID)
-		return true
 	}
-	return false
+	g.pamSessionsMu.Unlock()
+	if removed != nil {
+		close(removed.done)
+	}
+	return isLast
+}
+
+// Cancels prior entries and waits for them to clean up. RDP needs serial
+// bridges so drain writes don't interleave into the recording file.
+func (g *Gateway) evictExistingPAMSessions(sessionID string, timeout time.Duration) {
+	g.pamSessionsMu.Lock()
+	prior := g.pamSessions[sessionID]
+	g.pamSessionsMu.Unlock()
+	if len(prior) == 0 {
+		return
+	}
+	log.Info().Str("sessionId", sessionID).Int("priorCount", len(prior)).
+		Msg("Evicting existing PAM connections before starting new RDP bridge")
+	for _, e := range prior {
+		_ = e.conn.Close()
+		e.cancel()
+	}
+	deadline := time.After(timeout)
+	for _, e := range prior {
+		select {
+		case <-e.done:
+		case <-deadline:
+			log.Warn().Str("sessionId", sessionID).
+				Msg("Timed out waiting for prior PAM connection to clean up; proceeding anyway")
+			return
+		}
+	}
 }
 
 // CancelPAMSession kills all active connections for a PAM session
@@ -222,6 +255,7 @@ func (g *Gateway) CancelPAMSession(sessionID string) bool {
 	for _, e := range entries {
 		e.conn.Close()
 		e.cancel()
+		close(e.done)
 	}
 	g.closeMongoProxy(sessionID)
 	return true
@@ -833,6 +867,11 @@ func (g *Gateway) handleIncomingChannel(newChannel ssh.NewChannel) {
 		}
 		return
 	} else if forwardConfig.Mode == ForwardModePAM {
+		// RDP only: prior bridge must fully tear down before the new one starts,
+		// else overlapping drains write non-monotonic elapsedMs to the recording.
+		if forwardConfig.PAMConfig.ResourceType == session.ResourceTypeWindows {
+			g.evictExistingPAMSessions(forwardConfig.PAMConfig.SessionId, 5*time.Second)
+		}
 		sessionCtx, sessionCancel := context.WithCancel(g.ctx)
 		touchSession := g.RegisterPAMSession(forwardConfig.PAMConfig.SessionId, sessionCancel, tlsConn)
 		forwardConfig.PAMConfig.OnActivity = touchSession
@@ -844,7 +883,11 @@ func (g *Gateway) handleIncomingChannel(newChannel ssh.NewChannel) {
 			}
 		}
 		sessionCancel()
-		if lastConn := g.DeregisterPAMSession(forwardConfig.PAMConfig.SessionId, tlsConn); lastConn {
+		// RDP reconnects via a stable .rdp file within the session's validity
+		// window; terminating on disconnect would break that. Idle reaper /
+		// expiry / explicit cancel still end the session normally.
+		isRDP := forwardConfig.PAMConfig.ResourceType == session.ResourceTypeWindows
+		if lastConn := g.DeregisterPAMSession(forwardConfig.PAMConfig.SessionId, tlsConn); lastConn && !isRDP {
 			if err := forwardConfig.PAMConfig.SessionUploader.CleanupPAMSession(
 				forwardConfig.PAMConfig.SessionId, "connection_closed",
 			); err != nil {

packages/pam/handlers/rdp/bridge.go

@@ -15,3 +15,44 @@ type Bridge struct {
 	handle  uint64
 	cleanup func()
 }
+
+// EventType discriminates the variants in Event.
+type EventType uint8
+
+const (
+	EventTypeKeyboard    EventType = 1
+	EventTypeUnicode     EventType = 2
+	EventTypeMouse       EventType = 3
+	EventTypeTargetFrame EventType = 4
+)
+
+// Action identifies the RDP framing of a TargetFrame event.
+type Action uint8
+
+const (
+	ActionX224     Action = 0
+	ActionFastPath Action = 1
+)
+
+// Fields are reused across variants; switch on Type.
+type Event struct {
+	Type       EventType
+	ElapsedNs  uint64
+	Scancode   uint8
+	CodePoint  uint16
+	X          uint16
+	Y          uint16
+	Flags      uint32
+	WheelDelta int32
+	Action     Action
+	Payload    []byte
+}
+
+// PollResult discriminates PollEvent outcomes.
+type PollResult uint8
+
+const (
+	PollOK      PollResult = 0
+	PollTimeout PollResult = 1
+	PollEnded   PollResult = 2
+)

packages/pam/handlers/rdp/bridge_cgo_shared.go

@@ -5,6 +5,7 @@ package rdp
 /*
 #cgo CFLAGS: -I${SRCDIR}/native/include
 
+#include <stdlib.h>
 #include "rdp_bridge.h"
 */
 import "C"
@@ -14,6 +15,8 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"time"
+	"unsafe"
 )
 
 func (p *RDPProxy) HandleConnection(ctx context.Context, clientConn net.Conn) error {
@@ -37,16 +40,40 @@ func (p *RDPProxy) HandleConnection(ctx context.Context, clientConn net.Conn) er
 	}
 	defer bridge.Close()
 
+	// Drain bridge tap events into the session logger. The Rust side closes
+	// the events channel when the session ends, so the goroutine exits via
+	// PollEnded without needing an explicit shutdown signal.
+	drainCtx, cancelDrain := context.WithCancel(ctx)
+	drainDone := make(chan struct{})
+	go func() {
+		defer close(drainDone)
+		drainBridgeEvents(drainCtx, bridge, p.config.SessionLogger, p.config.SessionID, p.config.PriorElapsedNs, p.config.SessionUploader)
+	}()
+	// Wait for the drain to finish naturally on the normal-end path so the
+	// tail of the recording isn't dropped: PollEnded fires after the Rust
+	// side closes the events channel (post bridge.Wait return). Cancellation
+	// paths trigger cancelDrain() explicitly below to bail early.
+	defer func() {
+		select {
+		case <-drainDone:
+		case <-time.After(2 * pollTimeout):
+		}
+		// Always release the drain context (no-op if already cancelled).
+		cancelDrain()
+	}()
+
 	waitErr := make(chan error, 1)
 	go func() { waitErr <- bridge.Wait() }()
 
 	select {
 	case err := <-waitErr:
 		if err != nil && !errors.Is(err, ErrInvalidHandle) {
+			cancelDrain()
 			return fmt.Errorf("rdp proxy: session: %w", err)
 		}
 		return nil
 	case <-ctx.Done():
+		cancelDrain()
 		_ = bridge.Cancel()
 		<-waitErr
 		return ctx.Err()
@@ -90,8 +117,62 @@ func (b *Bridge) Close() error {
 	return nil
 }
 
-// IsSupported reports whether this build has a real RDP bridge. Used
-// by the gateway to decide whether to advertise RDP in the capabilities
-// response: a stub-build gateway that advertises support would route
-// RDP sessions only to fail them at connect time.
+// True when the real bridge is compiled in (vs the stub).
 func IsSupported() bool { return true }
+
+// PollEvent drains one tap event with the given timeout. The returned Event
+// is only meaningful when result == PollOK. PollEvent is not safe to call
+// concurrently for the same Bridge; serialize calls in a single goroutine.
+func (b *Bridge) PollEvent(timeout time.Duration) (PollResult, Event, error) {
+	timeoutMs := timeout.Milliseconds()
+	if timeoutMs < 0 {
+		timeoutMs = 0
+	}
+	if timeoutMs > int64(^C.uint32_t(0)) {
+		timeoutMs = int64(^C.uint32_t(0))
+	}
+
+	var raw C.struct_RdpEvent
+	rc := C.rdp_bridge_poll_event(C.uint64_t(b.handle), &raw, C.uint32_t(timeoutMs))
+
+	switch rc {
+	case C.RDP_POLL_OK:
+		// fall through to event materialization below
+	case C.RDP_POLL_TIMEOUT:
+		return PollTimeout, Event{}, nil
+	case C.RDP_POLL_ENDED:
+		return PollEnded, Event{}, nil
+	case C.RDP_POLL_INVALID_HANDLE:
+		return PollEnded, Event{}, ErrInvalidHandle
+	default:
+		return PollEnded, Event{}, fmt.Errorf("rdp bridge: poll returned unexpected status %d", int32(rc))
+	}
+
+	ev := Event{
+		Type:       EventType(uint8(raw.event_type)),
+		ElapsedNs:  uint64(raw.elapsed_ns),
+		Flags:      uint32(raw.flags),
+		WheelDelta: int32(raw.wheel_delta),
+		Action:     Action(uint8(raw.action)),
+	}
+	switch ev.Type {
+	case EventTypeKeyboard:
+		ev.Scancode = uint8(raw.value_a)
+	case EventTypeUnicode:
+		ev.CodePoint = uint16(raw.value_a)
+	case EventTypeMouse:
+		ev.X = uint16(raw.value_a)
+		ev.Y = uint16(raw.value_b)
+	case EventTypeTargetFrame:
+		// Always free the libc-malloc'd buffer Rust handed us, even if
+		// the copy below is empty -- ownership transfer is unconditional.
+		if raw.payload_ptr != nil {
+			defer C.free(unsafe.Pointer(raw.payload_ptr))
+			if raw.payload_len > 0 {
+				ev.Payload = C.GoBytes(unsafe.Pointer(raw.payload_ptr), C.int(raw.payload_len))
+			}
+		}
+	}
+
+	return PollOK, ev, nil
+}

packages/pam/handlers/rdp/bridge_cgo_unix.go

@@ -72,19 +72,8 @@ func startWithDupedFD(dupFd int, targetHost string, targetPort uint16, username,
 	return &Bridge{handle: uint64(handle)}, nil
 }
 
-// StartWithReadWriter adapts an fd-less Go byte stream (e.g. *tls.Conn
-// from the gateway's mTLS-wrapped virtual connection) to the bridge,
-// which needs a real file descriptor because the Rust side uses tokio's
-// TcpStream::from_raw_fd and does direct async I/O on the socket.
-//
-// Trick: open a loopback TCP pair. Hand one end's fd to the bridge (it
-// thinks it has a real client). Keep the other end in Go and shuttle
-// bytes between it and rw with two io.Copy goroutines.
-//
-//	rw (e.g. *tls.Conn)  <-io.Copy->  peer  <-kernel loopback->  accepted (fd -> Rust bridge)
-//
-// Cost: two extra in-process copies and a loopback round-trip per byte.
-// Negligible vs. the TLS + CredSSP work on either side.
+// Adapts an fd-less Go byte stream to the Rust bridge (which needs a real fd
+// for tokio's TcpStream::from_raw_fd) by routing through a loopback TCP pair.
 func StartWithReadWriter(rw io.ReadWriter, targetHost string, targetPort uint16, username, password, domain string) (*Bridge, error) {
 	listener, err := net.Listen("tcp", "127.0.0.1:0")
 	if err != nil {

packages/pam/handlers/rdp/bridge_stub.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"io"
 	"net"
+	"time"
 )
 
 // Stub implementations for builds without `-tags rdp` or on platforms
@@ -29,6 +30,10 @@ func (b *Bridge) Wait() error   { return ErrRdpUnavailable }
 func (b *Bridge) Cancel() error { return ErrRdpUnavailable }
 func (b *Bridge) Close() error  { return ErrRdpUnavailable }
 
+func (b *Bridge) PollEvent(_ time.Duration) (PollResult, Event, error) {
+	return PollEnded, Event{}, ErrRdpUnavailable
+}
+
 // IsSupported reports whether this build has a real RDP bridge. See the
 // rdp-enabled counterpart in bridge_cgo_shared.go for details.
 func IsSupported() bool { return false }

packages/pam/handlers/rdp/native/Cargo.toml

@@ -13,10 +13,12 @@ path = "src/lib.rs"
 [dependencies]
 ironrdp-acceptor = "0.8"
 ironrdp-connector = "0.8"
+ironrdp-core = "0.1"
 ironrdp-tokio = { version = "0.8", features = ["reqwest"] }
 ironrdp-pdu = "0.7"
 ironrdp-tls = { version = "0.2", features = ["rustls"] }
 x509-cert = { version = "0.2", features = ["std"] }
+libc = "0.2"
 
 tokio = { version = "1", features = ["full"] }
 tokio-util = "0.7"