Emit port-forward event only the first time, add port-forward-end eve…

…nt. Include cluster and pod metadta in event.

api/proto/teleport/legacy/types/events/events.proto

@@ -1617,6 +1617,21 @@ message PortForward {
 
   // Addr is a target port forwarding address
   string Addr = 5 [(gogoproto.jsontag) = "addr"];
+
+  // KubernetesCluster has information about a kubernetes cluster, if
+  // applicable.
+  KubernetesClusterMetadata KubernetesCluster = 6 [
+    (gogoproto.nullable) = false,
+    (gogoproto.embed) = true,
+    (gogoproto.jsontag) = ""
+  ];
+
+  // KubernetesPod has information about a kubernetes pod, if applicable.
+  KubernetesPodMetadata KubernetesPod = 7 [
+    (gogoproto.nullable) = false,
+    (gogoproto.embed) = true,
+    (gogoproto.jsontag) = ""
+  ];
 }
 
 // X11Forward is emitted when a user requests X11 protocol forwarding

lib/events/api.go

@@ -792,35 +792,35 @@ const (
 
 	// IntegrationCreateEvent is emitted when an integration resource is created.
 	IntegrationCreateEvent = "integration.create"
-	//IntegrationUpdateEvent is emitted when an integration resource is updated.
+	// IntegrationUpdateEvent is emitted when an integration resource is updated.
 	IntegrationUpdateEvent = "integration.update"
 	// IntegrationDeleteEvent is emitted when an integration resource is deleted.
 	IntegrationDeleteEvent = "integration.delete"
 
 	// PluginCreateEvent is emitted when a plugin resource is created.
 	PluginCreateEvent = "plugin.create"
-	//PluginUpdateEvent is emitted when a plugin resource is updated.
+	// PluginUpdateEvent is emitted when a plugin resource is updated.
 	PluginUpdateEvent = "plugin.update"
 	// PluginDeleteEvent is emitted when a plugin resource is deleted.
 	PluginDeleteEvent = "plugin.delete"
 
 	// StaticHostUserCreateEvent is emitted when a static host user resource is created.
 	StaticHostUserCreateEvent = "static_host_user.create"
-	//StaticHostUserUpdateEvent is emitted when a static host user resource is updated.
+	// StaticHostUserUpdateEvent is emitted when a static host user resource is updated.
 	StaticHostUserUpdateEvent = "static_host_user.update"
 	// StaticHostUserDeleteEvent is emitted when a static host user resource is deleted.
 	StaticHostUserDeleteEvent = "static_host_user.delete"
 
 	// CrownJewelCreateEvent is emitted when a crown jewel resource is created.
 	CrownJewelCreateEvent = "access_graph.crown_jewel.create"
-	//CrownJewelUpdateEvent is emitted when a crown jewel resource is updated.
+	// CrownJewelUpdateEvent is emitted when a crown jewel resource is updated.
 	CrownJewelUpdateEvent = "access_graph.crown_jewel.update"
 	// CrownJewelDeleteEvent is emitted when a crown jewel resource is deleted.
 	CrownJewelDeleteEvent = "access_graph.crown_jewel.delete"
 
 	// UserTaskCreateEvent is emitted when a user task resource is created.
 	UserTaskCreateEvent = "user_task.create"
-	//UserTaskUpdateEvent is emitted when a user task resource is updated.
+	// UserTaskUpdateEvent is emitted when a user task resource is updated.
 	UserTaskUpdateEvent = "user_task.update"
 	// UserTaskDeleteEvent is emitted when a user task resource is deleted.
 	UserTaskDeleteEvent = "user_task.delete"
@@ -868,15 +868,13 @@ const (
 	V3 = 3
 )
 
-var (
-	// SessionRecordingEvents is a list of events that are related to session
-	// recorings.
-	SessionRecordingEvents = []string{
-		SessionEndEvent,
-		WindowsDesktopSessionEndEvent,
-		DatabaseSessionEndEvent,
-	}
-)
+// SessionRecordingEvents is a list of events that are related to session
+// recorings.
+var SessionRecordingEvents = []string{
+	SessionEndEvent,
+	WindowsDesktopSessionEndEvent,
+	DatabaseSessionEndEvent,
+}
 
 // ServerMetadataGetter represents interface
 // that provides information about its server id

lib/events/codes.go

@@ -286,6 +286,8 @@ const (
 	ExecFailureCode = "T3002E"
 	// PortForwardCode is the port forward event code.
 	PortForwardCode = "T3003I"
+	// PortForwardStopCode is the port forward stop event code.
+	PortForwardStopCode = "T3003S"
 	// PortForwardFailureCode is the port forward failure event code.
 	PortForwardFailureCode = "T3003E"
 	// SCPDownloadCode is the file download event code.

lib/kube/proxy/forwarder.go

@@ -1774,10 +1774,12 @@ func (f *Forwarder) portForward(authCtx *authContext, w http.ResponseWriter, req
 		return nil, trace.Wrap(err)
 	}
 
+	auditSent := map[string]bool{} // Set of `addr`. Can be multiple ports on single call. Using bool to simplify the check.
 	onPortForward := func(addr string, success bool) {
-		if !sess.isLocalKubernetesCluster {
+		if !sess.isLocalKubernetesCluster || auditSent[addr] {
 			return
 		}
+		auditSent[addr] = true
 		portForward := &apievents.PortForward{
 			Metadata: apievents.Metadata{
 				Type: events.PortForwardEvent,
@@ -1793,6 +1795,11 @@ func (f *Forwarder) portForward(authCtx *authContext, w http.ResponseWriter, req
 			Status: apievents.Status{
 				Success: success,
 			},
+			KubernetesClusterMetadata: sess.eventClusterMeta(req),
+			KubernetesPodMetadata: apievents.KubernetesPodMetadata{
+				KubernetesPodNamespace: p.ByName("podNamespace"),
+				KubernetesPodName:      p.ByName("podName"),
+			},
 		}
 		if !success {
 			portForward.Code = events.PortForwardFailureCode
@@ -1801,6 +1808,31 @@ func (f *Forwarder) portForward(authCtx *authContext, w http.ResponseWriter, req
 			f.log.WithError(err).Warn("Failed to emit event.")
 		}
 	}
+	defer func() {
+		for addr := range auditSent {
+			portForward := &apievents.PortForward{
+				Metadata: apievents.Metadata{
+					Type: events.PortForwardEvent,
+					Code: events.PortForwardStopCode,
+				},
+				UserMetadata: authCtx.eventUserMeta(),
+				ConnectionMetadata: apievents.ConnectionMetadata{
+					LocalAddr:  sess.kubeAddress,
+					RemoteAddr: req.RemoteAddr,
+					Protocol:   events.EventProtocolKube,
+				},
+				Addr:                      addr,
+				KubernetesClusterMetadata: sess.eventClusterMeta(req),
+				KubernetesPodMetadata: apievents.KubernetesPodMetadata{
+					KubernetesPodNamespace: p.ByName("podNamespace"),
+					KubernetesPodName:      p.ByName("podName"),
+				},
+			}
+			if err := f.cfg.Emitter.EmitAuditEvent(f.ctx, portForward); err != nil {
+				f.log.WithError(err).Warn("Failed to emit event.")
+			}
+		}
+	}()
 
 	q := req.URL.Query()
 	request := portForwardRequest{

web/packages/teleport/src/Audit/EventList/EventTypeCell.tsx

@@ -195,6 +195,7 @@ const EventIconMap: Record<EventCode, any> = {
   [eventCodes.CLIENT_DISCONNECT]: Icons.Info,
   [eventCodes.PORTFORWARD]: Icons.Info,
   [eventCodes.PORTFORWARD_FAILURE]: Icons.Info,
+  [eventCodes.PORTFORWARD_STOP]: Icons.Info,
   [eventCodes.SUBSYSTEM]: Icons.Info,
   [eventCodes.SUBSYSTEM_FAILURE]: Icons.Info,
   [eventCodes.LOCK_CREATED]: Icons.Lock,

web/packages/teleport/src/services/audit/makeEvent.ts

@@ -231,6 +231,11 @@ export const formatters: Formatters = {
     format: ({ user, error }) =>
       `User [${user}] port forwarding request failed: ${error}`,
   },
+  [eventCodes.PORTFORWARD_STOP]: {
+    type: 'port',
+    desc: 'Port Forwarding Stopped',
+    format: ({ user }) => `User [${user}] stopped port forwarding`,
+  },
   [eventCodes.SAML_CONNECTOR_CREATED]: {
     type: 'saml.created',
     desc: 'SAML Connector Created',

web/packages/teleport/src/services/audit/types.ts

@@ -143,6 +143,7 @@ export const eventCodes = {
   OIDC_CONNECTOR_DELETED: 'T8101I',
   OIDC_CONNECTOR_UPDATED: 'T8102I',
   PORTFORWARD_FAILURE: 'T3003E',
+  PORTFORWARD_STOP: 'T3003S',
   PORTFORWARD: 'T3003I',
   RECOVERY_TOKEN_CREATED: 'T6001I',
   PRIVILEGE_TOKEN_CREATED: 'T6002I',
@@ -394,6 +395,7 @@ export type RawEvents = {
     typeof eventCodes.OIDC_CONNECTOR_UPDATED
   >;
   [eventCodes.PORTFORWARD]: RawEvent<typeof eventCodes.PORTFORWARD>;
+  [eventCodes.PORTFORWARD_STOP]: RawEvent<typeof eventCodes.PORTFORWARD_STOP>;
   [eventCodes.PORTFORWARD_FAILURE]: RawEvent<
     typeof eventCodes.PORTFORWARD_FAILURE,
     {