[v16] Improving regular SSH port forwarding audit logs

lib/events/api.go

@@ -262,10 +262,13 @@ const (
 	X11ForwardErr     = "error"
 
 	// Port forwarding event
-	PortForwardEvent   = "port"
-	PortForwardAddr    = "addr"
-	PortForwardSuccess = "success"
-	PortForwardErr     = "error"
+	PortForwardEvent           = "port"
+	PortForwardLocalEvent      = "port.local"
+	PortForwardRemoteEvent     = "port.remote"
+	PortForwardRemoteConnEvent = "port.remote_conn"
+	PortForwardAddr            = "addr"
+	PortForwardSuccess         = "success"
+	PortForwardErr             = "error"
 
 	// AuthAttemptEvent is authentication attempt that either
 	// succeeded or failed based on event status

lib/events/dynamic.go

@@ -107,6 +107,12 @@ func FromEventFields(fields EventFields) (events.AuditEvent, error) {
 		e = &events.X11Forward{}
 	case PortForwardEvent:
 		e = &events.PortForward{}
+	case PortForwardLocalEvent:
+		e = &events.PortForward{}
+	case PortForwardRemoteEvent:
+		e = &events.PortForward{}
+	case PortForwardRemoteConnEvent:
+		e = &events.PortForward{}
 	case AuthAttemptEvent:
 		e = &events.AuthAttempt{}
 	case SCPEvent:

lib/srv/ctx.go

@@ -1401,19 +1401,19 @@ func (c *ServerContext) GetSessionMetadata() apievents.SessionMetadata {
 	}
 }
 
-func (c *ServerContext) GetPortForwardEvent() apievents.PortForward {
+func (c *ServerContext) GetPortForwardEvent(evType, code, addr string) apievents.PortForward {
 	sconn := c.ConnectionContext.ServerConn
 	return apievents.PortForward{
 		Metadata: apievents.Metadata{
-			Type: events.PortForwardEvent,
-			Code: events.PortForwardCode,
+			Type: evType,
+			Code: code,
 		},
 		UserMetadata: c.Identity.GetUserMetadata(),
 		ConnectionMetadata: apievents.ConnectionMetadata{
 			LocalAddr:  sconn.LocalAddr().String(),
 			RemoteAddr: sconn.RemoteAddr().String(),
 		},
-		Addr: c.DstAddr,
+		Addr: addr,
 		Status: apievents.Status{
 			Success: true,
 		},

lib/srv/forward/sshserver.go

@@ -930,7 +930,7 @@ func (s *Server) handleForwardedTCPIPRequest(ctx context.Context, nch ssh.NewCha
 	go io.Copy(io.Discard, ch.Stderr())
 	ch = scx.TrackActivity(ch)
 
-	event := scx.GetPortForwardEvent()
+	event := scx.GetPortForwardEvent(events.PortForwardEvent, events.PortForwardCode, scx.DstAddr)
 	if err := s.EmitAuditEvent(ctx, &event); err != nil {
 		s.log.WithError(err).Error("Failed to emit audit event.")
 	}
@@ -1095,7 +1095,7 @@ func (s *Server) handleDirectTCPIPRequest(ctx context.Context, ch ssh.Channel, r
 	}
 	defer conn.Close()
 
-	event := scx.GetPortForwardEvent()
+	event := scx.GetPortForwardEvent(events.PortForwardEvent, events.PortForwardFailureCode, scx.DstAddr)
 	if err := s.EmitAuditEvent(s.closeContext, &event); err != nil {
 		scx.WithError(err).Warn("Failed to emit port forward event.")
 	}

lib/srv/regular/sshserver.go

@@ -1688,27 +1688,17 @@ func (s *Server) handleDirectTCPIPRequest(ctx context.Context, ccx *sshutils.Con
 		return
 	}
 
+	startEvent := scx.GetPortForwardEvent(events.PortForwardLocalEvent, events.PortForwardCode, scx.DstAddr)
+	s.emitAuditEventWithLog(ctx, &startEvent)
+
 	if err := utils.ProxyConn(ctx, conn, channel); err != nil && !errors.Is(err, io.EOF) && !errors.Is(err, os.ErrClosed) {
-		s.Logger.Warnf("Connection problem in direct-tcpip channel: %v %T.", trace.DebugReport(err), err)
+		errEvent := scx.GetPortForwardEvent(events.PortForwardLocalEvent, events.PortForwardFailureCode, scx.DstAddr)
+		s.emitAuditEventWithLog(ctx, &errEvent)
+		slog.WarnContext(ctx, "Connection problem in direct-tcpip channel", "error", err)
 	}
 
-	if err := s.EmitAuditEvent(s.ctx, &apievents.PortForward{
-		Metadata: apievents.Metadata{
-			Type: events.PortForwardEvent,
-			Code: events.PortForwardCode,
-		},
-		UserMetadata: scx.Identity.GetUserMetadata(),
-		ConnectionMetadata: apievents.ConnectionMetadata{
-			LocalAddr:  scx.ServerConn.LocalAddr().String(),
-			RemoteAddr: scx.ServerConn.RemoteAddr().String(),
-		},
-		Addr: scx.DstAddr,
-		Status: apievents.Status{
-			Success: true,
-		},
-	}); err != nil {
-		s.Logger.WithError(err).Warn("Failed to emit port forward event.")
-	}
+	stopEvent := scx.GetPortForwardEvent(events.PortForwardLocalEvent, events.PortForwardStopCode, scx.DstAddr)
+	s.emitAuditEventWithLog(ctx, &stopEvent)
 }
 
 // handleSessionRequests handles out of band session requests once the session
@@ -2042,9 +2032,7 @@ func (s *Server) handleX11Forward(ch ssh.Channel, req *ssh.Request, ctx *srv.Ser
 			s.replyError(ch, req, err)
 			err = nil
 		}
-		if err := s.EmitAuditEvent(s.ctx, event); err != nil {
-			s.Logger.WithError(err).Warn("Failed to emit x11-forward event.")
-		}
+		s.emitAuditEventWithLog(s.ctx, event)
 	}()
 
 	// check if X11 forwarding is disabled, or if xauth can't be handled.
@@ -2321,6 +2309,7 @@ func (s *Server) createForwardingContext(ctx context.Context, ccx *sshutils.Conn
 	if err != nil {
 		return nil, nil, trace.Wrap(err)
 	}
+
 	listenAddr := sshutils.JoinHostPort(req.Addr, req.Port)
 	scx.IsTestStub = s.isTestStub
 	scx.ExecType = teleport.TCPIPForwardRequest
@@ -2359,13 +2348,72 @@ func (s *Server) handleTCPIPForwardRequest(ctx context.Context, ccx *sshutils.Co
 	}
 	scx.SrcAddr = sshutils.JoinHostPort(srcHost, listenPort)
 
-	event := scx.GetPortForwardEvent()
-	if err := s.EmitAuditEvent(ctx, &event); err != nil {
-		s.Logger.WithError(err).Warn("Failed to emit audit event.")
-	}
-	if err := sshutils.StartRemoteListener(ctx, scx.ConnectionContext.ServerConn, scx.SrcAddr, listener); err != nil {
-		return trace.Wrap(err)
-	}
+	event := scx.GetPortForwardEvent(events.PortForwardRemoteEvent, events.PortForwardCode, scx.SrcAddr)
+	s.emitAuditEventWithLog(ctx, &event)
+
+	// spawn remote forwarding handler to multiplex connections to the forwarded port
+	go func() {
+		stopEvent := scx.GetPortForwardEvent(events.PortForwardRemoteEvent, events.PortForwardStopCode, scx.SrcAddr)
+		defer s.emitAuditEventWithLog(ctx, &stopEvent)
+
+		for {
+			conn, err := listener.Accept()
+			if err != nil {
+				if !utils.IsOKNetworkError(err) {
+					slog.WarnContext(ctx, "failed to accept connection", "error", err)
+				}
+				return
+			}
+			logger := slog.With(
+				"src_addr", scx.SrcAddr,
+				"remote_addr", conn.RemoteAddr().String(),
+			)
+
+			dstHost, dstPort, err := sshutils.SplitHostPort(conn.RemoteAddr().String())
+			if err != nil {
+				conn.Close()
+				logger.WarnContext(ctx, "failed to parse addr", "error", err)
+				return
+			}
+
+			req := sshutils.ForwardedTCPIPRequest{
+				Addr:     srcHost,
+				Port:     listenPort,
+				OrigAddr: dstHost,
+				OrigPort: dstPort,
+			}
+			if err := req.CheckAndSetDefaults(); err != nil {
+				conn.Close()
+				logger.WarnContext(ctx, "failed to create forwarded tcpip request", "error", err)
+				return
+			}
+			reqBytes := ssh.Marshal(req)
+
+			ch, rch, err := scx.ConnectionContext.ServerConn.OpenChannel(teleport.ChanForwardedTCPIP, reqBytes)
+			if err != nil {
+				conn.Close()
+				logger.WarnContext(ctx, "failed to open channel", "error", err)
+				continue
+			}
+			go ssh.DiscardRequests(rch)
+			go io.Copy(io.Discard, ch.Stderr())
+			go func() {
+				startEvent := scx.GetPortForwardEvent(events.PortForwardRemoteConnEvent, events.PortForwardCode, scx.SrcAddr)
+				startEvent.RemoteAddr = conn.RemoteAddr().String()
+				s.emitAuditEventWithLog(ctx, &startEvent)
+
+				if err := utils.ProxyConn(ctx, conn, ch); err != nil {
+					errEvent := scx.GetPortForwardEvent(events.PortForwardRemoteConnEvent, events.PortForwardFailureCode, scx.SrcAddr)
+					errEvent.RemoteAddr = conn.RemoteAddr().String()
+					s.emitAuditEventWithLog(ctx, &errEvent)
+				}
+
+				stopEvent := scx.GetPortForwardEvent(events.PortForwardRemoteConnEvent, events.PortForwardStopCode, scx.SrcAddr)
+				stopEvent.RemoteAddr = conn.RemoteAddr().String()
+				s.emitAuditEventWithLog(ctx, &stopEvent)
+			}()
+		}
+	}()
 
 	// Report addr back to the client.
 	if r.WantReply {
@@ -2397,14 +2445,14 @@ func (s *Server) handleCancelTCPIPForwardRequest(ctx context.Context, ccx *sshut
 		return trace.Wrap(err)
 	}
 	defer scx.Close()
-
 	listener, ok := s.remoteForwardingMap.LoadAndDelete(scx.SrcAddr)
 	if !ok {
 		return trace.NotFound("no remote forwarding listener at %v", scx.SrcAddr)
 	}
 	if err := r.Reply(true, nil); err != nil {
 		s.Logger.Warnf("Failed to reply to %q request: %v", r.Type, err)
 	}
+
 	return trace.Wrap(listener.Close())
 }
 
@@ -2447,7 +2495,7 @@ func (s *Server) parseSubsystemRequest(req *ssh.Request, ctx *srv.ServerContext)
 	case r.Name == teleport.SFTPSubsystem:
 		err := ctx.CheckSFTPAllowed(s.reg)
 		if err != nil {
-			s.EmitAuditEvent(context.Background(), &apievents.SFTP{
+			s.emitAuditEventWithLog(context.Background(), &apievents.SFTP{
 				Metadata: apievents.Metadata{
 					Code: events.SFTPDisallowedCode,
 					Type: events.SFTPEvent,
@@ -2494,3 +2542,9 @@ func (s *Server) handlePuTTYWinadj(ch ssh.Channel, req *ssh.Request) error {
 	req.WantReply = false
 	return nil
 }
+
+func (s *Server) emitAuditEventWithLog(ctx context.Context, event apievents.AuditEvent) {
+	if err := s.EmitAuditEvent(ctx, event); err != nil {
+		slog.WarnContext(ctx, "Failed to emit event", "type", event.GetType(), "code", event.GetCode())
+	}
+}

lib/sshutils/mock.go

@@ -22,6 +22,7 @@ import (
 	"errors"
 	"io"
 
+	"github.com/stretchr/testify/mock"
 	"golang.org/x/crypto/ssh"
 )
 
@@ -60,10 +61,54 @@ func (mc *mockChannel) Stderr() io.ReadWriter {
 	return fakeReaderWriter{}
 }
 
-type mockSSHConn struct {
-	mockChan *mockChannel
+type mockSSHNewChannel struct {
+	mock.Mock
+	ssh.NewChannel
 }
 
-func (mc *mockSSHConn) OpenChannel(name string, data []byte) (ssh.Channel, <-chan *ssh.Request, error) {
-	return mc.mockChan, make(<-chan *ssh.Request), nil
+func newMockSSHNewChannel(channelType string) *mockSSHNewChannel {
+	m := new(mockSSHNewChannel)
+	m.On("ChannelType").Return(channelType)
+	m.On("Reject", mock.Anything, mock.Anything).Return(nil)
+	return m
+}
+
+func (m *mockSSHNewChannel) ChannelType() string {
+	return m.Called().String(0)
+}
+
+func (m *mockSSHNewChannel) Reject(reason ssh.RejectionReason, message string) error {
+	args := m.Called(reason, message)
+	return args.Error(0)
+}
+
+type mockSSHChannel struct {
+	mock.Mock
+	ssh.Channel
+}
+
+func newMockSSHChannel() *mockSSHChannel {
+	m := new(mockSSHChannel)
+	m.On("SendRequest", mock.Anything, mock.Anything, mock.Anything).Return(false, nil)
+	return m
+}
+
+func (m *mockSSHChannel) SendRequest(name string, wantReply bool, payload []byte) (bool, error) {
+	args := m.Called(name, wantReply, payload)
+	return args.Bool(0), args.Error(1)
+}
+
+type mockSSHRequest struct {
+	mock.Mock
+}
+
+func newMockSSHRequest() *mockSSHRequest {
+	m := new(mockSSHRequest)
+	m.On("Reply", mock.Anything, mock.Anything).Return(nil)
+	return m
+}
+
+func (m *mockSSHRequest) Reply(ok bool, payload []byte) error {
+	args := m.Called(ok, payload)
+	return args.Error(0)
 }

lib/sshutils/tcpip.go

@@ -19,16 +19,9 @@
 package sshutils
 
 import (
-	"context"
-	"io"
-	"net"
-
 	"github.com/gravitational/trace"
 	log "github.com/sirupsen/logrus"
 	"golang.org/x/crypto/ssh"
-
-	"github.com/gravitational/teleport"
-	"github.com/gravitational/teleport/lib/utils"
 )
 
 // DirectTCPIPReq represents the payload of an SSH "direct-tcpip" or
@@ -72,64 +65,3 @@ func ParseTCPIPForwardReq(data []byte) (*TCPIPForwardReq, error) {
 	}
 	return &r, nil
 }
-
-type channelOpener interface {
-	OpenChannel(name string, data []byte) (ssh.Channel, <-chan *ssh.Request, error)
-}
-
-// StartRemoteListener listens on the given listener and forwards any accepted
-// connections over a new "forwarded-tcpip" channel.
-func StartRemoteListener(ctx context.Context, sshConn channelOpener, srcAddr string, listener net.Listener) error {
-	srcHost, srcPort, err := SplitHostPort(srcAddr)
-	if err != nil {
-		return trace.Wrap(err)
-	}
-
-	go func() {
-		for {
-			conn, err := listener.Accept()
-			if err != nil {
-				if !utils.IsOKNetworkError(err) {
-					log.WithError(err).Warn("failed to accept connection")
-				}
-				return
-			}
-			logger := log.WithFields(log.Fields{
-				"srcAddr":    srcAddr,
-				"remoteAddr": conn.RemoteAddr().String(),
-			})
-
-			dstHost, dstPort, err := SplitHostPort(conn.RemoteAddr().String())
-			if err != nil {
-				conn.Close()
-				logger.WithError(err).Warn("failed to parse addr")
-				return
-			}
-
-			req := ForwardedTCPIPRequest{
-				Addr:     srcHost,
-				Port:     srcPort,
-				OrigAddr: dstHost,
-				OrigPort: dstPort,
-			}
-			if err := req.CheckAndSetDefaults(); err != nil {
-				conn.Close()
-				logger.WithError(err).Warn("failed to create forwarded tcpip request")
-				return
-			}
-			reqBytes := ssh.Marshal(req)
-
-			ch, rch, err := sshConn.OpenChannel(teleport.ChanForwardedTCPIP, reqBytes)
-			if err != nil {
-				conn.Close()
-				logger.WithError(err).Warn("failed to open channel")
-				continue
-			}
-			go ssh.DiscardRequests(rch)
-			go io.Copy(io.Discard, ch.Stderr())
-			go utils.ProxyConn(ctx, conn, ch)
-		}
-	}()
-
-	return nil
-}