Audit Logging Using go-eventlogger Under Experiment Flag (#22234)

* add hashfunc field to EntryFormatter struct and adjust NewEntryFormatter function and tests
* add HeaderAdjuster interface and require it in EntryFormatter
* adjust all references to NewEntryFormatter to include a HeaderAdjuster parameter
* replace use of hash function in AuditedHeadersConfig's ApplyConfig method with Salter interface instance
* export audit.NewEvent function and adjust function signature
* add eventlogger based handling in LogRequest
* adjust eventlogger.Broker according to number of backends
* record auditing metrics
* only send events if a pipeline is registered
* remove TODO comments
* remove unused struct and method
* move setup of audited headers earlier into Core's initialization
* adjust entry_formatter to properly handle request headers
* protect against potential segmentation fault
* moved common code out of both switch cases
* protect against case where a.Data.Request or a.Data.Request.Headers is nil

---------

Co-authored-by: Peter Wilson <peter.wilson@hashicorp.com>

audit/entry_formatter.go

@@ -86,10 +86,28 @@ func (f *EntryFormatter) Process(ctx context.Context, e *eventlogger.Event) (*ev
 	}
 
 	var result []byte
+	data := new(logical.LogInput)
+	headers := make(map[string][]string)
+
+	if a.Data != nil {
+		*data = *a.Data
+		if a.Data.Request != nil && a.Data.Request.Headers != nil {
+			headers = a.Data.Request.Headers
+		}
+	}
+
+	if f.headerFormatter != nil {
+		adjustedHeaders, err := f.headerFormatter.ApplyConfig(ctx, headers, f.salter)
+		if err != nil {
+			return nil, fmt.Errorf("%s: unable to transform headers for auditing: %w", op, err)
+		}
+
+		data.Request.Headers = adjustedHeaders
+	}
 
 	switch a.Subtype {
 	case RequestType:
-		entry, err := f.FormatRequest(ctx, a.Data)
+		entry, err := f.FormatRequest(ctx, data)
 		if err != nil {
 			return nil, fmt.Errorf("%s: unable to parse request from audit event: %w", op, err)
 		}
@@ -99,7 +117,7 @@ func (f *EntryFormatter) Process(ctx context.Context, e *eventlogger.Event) (*ev
 			return nil, fmt.Errorf("%s: unable to format request: %w", op, err)
 		}
 	case ResponseType:
-		entry, err := f.FormatResponse(ctx, a.Data)
+		entry, err := f.FormatResponse(ctx, data)
 		if err != nil {
 			return nil, fmt.Errorf("%s: unable to parse response from audit event: %w", op, err)
 		}

audit/entry_formatter_test.go

@@ -25,7 +25,7 @@ func fakeEvent(tb testing.TB, subtype subtype, format format, input *logical.Log
 
 	date := time.Date(2023, time.July, 11, 15, 49, 10, 0o0, time.Local)
 
-	auditEvent, err := newEvent(subtype, format,
+	auditEvent, err := NewEvent(subtype,
 		WithID("123"),
 		WithNow(date),
 	)
@@ -33,7 +33,6 @@ func fakeEvent(tb testing.TB, subtype subtype, format format, input *logical.Log
 	require.NotNil(tb, auditEvent)
 	require.Equal(tb, "123", auditEvent.ID)
 	require.Equal(tb, "v0.1", auditEvent.Version)
-	require.Equal(tb, format, auditEvent.RequiredFormat)
 	require.Equal(tb, subtype, auditEvent.Subtype)
 	require.Equal(tb, date, auditEvent.Timestamp)
 

audit/event.go

@@ -9,11 +9,10 @@ import (
 	"github.com/hashicorp/vault/internal/observability/event"
 )
 
-// newEvent should be used to create an audit event.
-// subtype and format are needed for audit.
-// It will generate an ID if no ID is supplied.
-// Supported options: WithID, WithNow.
-func newEvent(s subtype, f format, opt ...Option) (*auditEvent, error) {
+// NewEvent should be used to create an audit event. The subtype field is needed
+// for audit events. It will generate an ID if no ID is supplied. Supported
+// options: WithID, WithNow.
+func NewEvent(s subtype, opt ...Option) (*auditEvent, error) {
 	const op = "audit.newEvent"
 
 	// Get the default options
@@ -32,11 +31,10 @@ func newEvent(s subtype, f format, opt ...Option) (*auditEvent, error) {
 	}
 
 	audit := &auditEvent{
-		ID:             opts.withID,
-		Timestamp:      opts.withNow,
-		Version:        version,
-		Subtype:        s,
-		RequiredFormat: f,
+		ID:        opts.withID,
+		Timestamp: opts.withNow,
+		Version:   version,
+		Subtype:   s,
 	}
 
 	if err := audit.validate(); err != nil {
@@ -70,11 +68,6 @@ func (a *auditEvent) validate() error {
 		return fmt.Errorf("%s: %w", op, err)
 	}
 
-	err = a.RequiredFormat.validate()
-	if err != nil {
-		return fmt.Errorf("%s: %w", op, err)
-	}
-
 	return nil
 }
 

audit/event_test.go

@@ -82,7 +82,7 @@ func TestAuditEvent_new(t *testing.T) {
 		t.Run(name, func(t *testing.T) {
 			t.Parallel()
 
-			audit, err := newEvent(tc.Subtype, tc.Format, tc.Options...)
+			audit, err := NewEvent(tc.Subtype, tc.Options...)
 			switch {
 			case tc.IsErrorExpected:
 				require.Error(t, err)
@@ -93,7 +93,6 @@ func TestAuditEvent_new(t *testing.T) {
 				require.NotNil(t, audit)
 				require.Equal(t, tc.ExpectedID, audit.ID)
 				require.Equal(t, tc.ExpectedSubtype, audit.Subtype)
-				require.Equal(t, tc.ExpectedFormat, audit.RequiredFormat)
 				switch {
 				case tc.IsNowExpected:
 					require.True(t, time.Now().After(audit.Timestamp))
@@ -125,72 +124,55 @@ func TestAuditEvent_Validate(t *testing.T) {
 		},
 		"id-empty": {
 			Value: &auditEvent{
-				ID:             "",
-				Version:        version,
-				Subtype:        RequestType,
-				Timestamp:      time.Now(),
-				Data:           nil,
-				RequiredFormat: JSONFormat,
+				ID:        "",
+				Version:   version,
+				Subtype:   RequestType,
+				Timestamp: time.Now(),
+				Data:      nil,
 			},
 			IsErrorExpected:      true,
 			ExpectedErrorMessage: "audit.(auditEvent).validate: missing ID: invalid parameter",
 		},
 		"version-fiddled": {
 			Value: &auditEvent{
-				ID:             "audit_123",
-				Version:        "magic-v2",
-				Subtype:        RequestType,
-				Timestamp:      time.Now(),
-				Data:           nil,
-				RequiredFormat: JSONFormat,
+				ID:        "audit_123",
+				Version:   "magic-v2",
+				Subtype:   RequestType,
+				Timestamp: time.Now(),
+				Data:      nil,
 			},
 			IsErrorExpected:      true,
 			ExpectedErrorMessage: "audit.(auditEvent).validate: event version unsupported: invalid parameter",
 		},
 		"subtype-fiddled": {
 			Value: &auditEvent{
-				ID:             "audit_123",
-				Version:        version,
-				Subtype:        subtype("moon"),
-				Timestamp:      time.Now(),
-				Data:           nil,
-				RequiredFormat: JSONFormat,
+				ID:        "audit_123",
+				Version:   version,
+				Subtype:   subtype("moon"),
+				Timestamp: time.Now(),
+				Data:      nil,
 			},
 			IsErrorExpected:      true,
 			ExpectedErrorMessage: "audit.(auditEvent).validate: audit.(subtype).validate: 'moon' is not a valid event subtype: invalid parameter",
 		},
-		"format-fiddled": {
-			Value: &auditEvent{
-				ID:             "audit_123",
-				Version:        version,
-				Subtype:        ResponseType,
-				Timestamp:      time.Now(),
-				Data:           nil,
-				RequiredFormat: format("blah"),
-			},
-			IsErrorExpected:      true,
-			ExpectedErrorMessage: "audit.(auditEvent).validate: audit.(format).validate: 'blah' is not a valid format: invalid parameter",
-		},
 		"default-time": {
 			Value: &auditEvent{
-				ID:             "audit_123",
-				Version:        version,
-				Subtype:        ResponseType,
-				Timestamp:      time.Time{},
-				Data:           nil,
-				RequiredFormat: JSONFormat,
+				ID:        "audit_123",
+				Version:   version,
+				Subtype:   ResponseType,
+				Timestamp: time.Time{},
+				Data:      nil,
 			},
 			IsErrorExpected:      true,
 			ExpectedErrorMessage: "audit.(auditEvent).validate: event timestamp cannot be the zero time instant: invalid parameter",
 		},
 		"valid": {
 			Value: &auditEvent{
-				ID:             "audit_123",
-				Version:        version,
-				Subtype:        ResponseType,
-				Timestamp:      time.Now(),
-				Data:           nil,
-				RequiredFormat: JSONFormat,
+				ID:        "audit_123",
+				Version:   version,
+				Subtype:   ResponseType,
+				Timestamp: time.Now(),
+				Data:      nil,
 			},
 			IsErrorExpected: false,
 		},

audit/types.go

@@ -37,12 +37,11 @@ type format string
 
 // auditEvent is the audit event.
 type auditEvent struct {
-	ID             string            `json:"id"`
-	Version        string            `json:"version"`
-	Subtype        subtype           `json:"subtype"` // the subtype of the audit event.
-	Timestamp      time.Time         `json:"timestamp"`
-	Data           *logical.LogInput `json:"data"`
-	RequiredFormat format            `json:"format"`
+	ID        string            `json:"id"`
+	Version   string            `json:"version"`
+	Subtype   subtype           `json:"subtype"` // the subtype of the audit event.
+	Timestamp time.Time         `json:"timestamp"`
+	Data      *logical.LogInput `json:"data"`
 }
 
 // Option is how options are passed as arguments.

internal/observability/event/event_type.go

@@ -35,3 +35,8 @@ func GenerateNodeID() (eventlogger.NodeID, error) {
 
 	return eventlogger.NodeID(id), err
 }
+
+// String returns the string version of an EventType.
+func (t EventType) String() string {
+	return string(t)
+}

vault/audit_broker.go

@@ -10,6 +10,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/hashicorp/vault/internal/observability/event"
+
 	metrics "github.com/armon/go-metrics"
 	"github.com/hashicorp/eventlogger"
 	log "github.com/hashicorp/go-hclog"
@@ -39,9 +41,6 @@ func NewAuditBroker(log log.Logger, useEventLogger bool) (*AuditBroker, error) {
 	var err error
 
 	if useEventLogger {
-		// Ignoring the second error return value since an error will only occur
-		// if an unrecognized eventlogger.RegistrationPolicy is provided to an
-		// eventlogger.Option function.
 		eventBroker, err = eventlogger.NewBroker(eventlogger.WithNodeRegistrationPolicy(eventlogger.DenyOverwrite), eventlogger.WithPipelineRegistrationPolicy(eventlogger.DenyOverwrite))
 		if err != nil {
 			return nil, fmt.Errorf("error creating event broker for audit events: %w", err)
@@ -61,18 +60,23 @@ func (a *AuditBroker) Register(name string, b audit.Backend, local bool) error {
 	a.Lock()
 	defer a.Unlock()
 
-	if a.broker != nil {
-		err := b.RegisterNodesAndPipeline(a.broker, name)
-		if err != nil {
-			return err
-		}
-	}
-
 	a.backends[name] = backendEntry{
 		backend: b,
 		local:   local,
 	}
 
+	if a.broker != nil {
+		err := a.broker.SetSuccessThresholdSinks(eventlogger.EventType(event.AuditType.String()), 1)
+		if err != nil {
+			return err
+		}
+
+		err = b.RegisterNodesAndPipeline(a.broker, name)
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -87,11 +91,18 @@ func (a *AuditBroker) Deregister(ctx context.Context, name string) error {
 	delete(a.backends, name)
 
 	if a.broker != nil {
+		if len(a.backends) == 0 {
+			err := a.broker.SetSuccessThresholdSinks(eventlogger.EventType(event.AuditType.String()), 0)
+			if err != nil {
+				return err
+			}
+		}
+
 		// The first return value, a bool, indicates whether
 		// RemovePipelineAndNodes encountered the error while evaluating
 		// pre-conditions (false) or once it started removing the pipeline and
 		// the nodes (true). This code doesn't care either way.
-		_, err := a.broker.RemovePipelineAndNodes(ctx, eventlogger.EventType("audit"), eventlogger.PipelineID(name))
+		_, err := a.broker.RemovePipelineAndNodes(ctx, eventlogger.EventType(event.AuditType.String()), eventlogger.PipelineID(name))
 		if err != nil {
 			return err
 		}
@@ -166,40 +177,57 @@ func (a *AuditBroker) LogRequest(ctx context.Context, in *logical.LogInput, head
 		metrics.IncrCounter([]string{"audit", "log_request_failure"}, failure)
 	}()
 
-	// All logged requests must have an identifier
-	//if req.ID == "" {
-	//	a.logger.Error("missing identifier in request object", "request_path", req.Path)
-	//	retErr = multierror.Append(retErr, fmt.Errorf("missing identifier in request object: %s", req.Path))
-	//	return
-	//}
-
 	headers := in.Request.Headers
 	defer func() {
 		in.Request.Headers = headers
 	}()
 
-	// Ensure at least one backend logs
-	anyLogged := false
-	for name, be := range a.backends {
-		in.Request.Headers = nil
-		transHeaders, thErr := headersConfig.ApplyConfig(ctx, headers, be.backend)
-		if thErr != nil {
-			a.logger.Error("backend failed to include headers", "backend", name, "error", thErr)
-			continue
-		}
-		in.Request.Headers = transHeaders
+	// Old behavior (no events)
+	if a.broker == nil {
+		// Ensure at least one backend logs
+		anyLogged := false
+		for name, be := range a.backends {
+			in.Request.Headers = nil
+			transHeaders, thErr := headersConfig.ApplyConfig(ctx, headers, be.backend)
+			if thErr != nil {
+				a.logger.Error("backend failed to include headers", "backend", name, "error", thErr)
+				continue
+			}
+			in.Request.Headers = transHeaders
 
-		start := time.Now()
-		lrErr := be.backend.LogRequest(ctx, in)
-		metrics.MeasureSince([]string{"audit", name, "log_request"}, start)
-		if lrErr != nil {
-			a.logger.Error("backend failed to log request", "backend", name, "error", lrErr)
-		} else {
-			anyLogged = true
+			start := time.Now()
+			lrErr := be.backend.LogRequest(ctx, in)
+			metrics.MeasureSince([]string{"audit", name, "log_request"}, start)
+			if lrErr != nil {
+				a.logger.Error("backend failed to log request", "backend", name, "error", lrErr)
+			} else {
+				anyLogged = true
+			}
+		}
+		if !anyLogged && len(a.backends) > 0 {
+			retErr = multierror.Append(retErr, fmt.Errorf("no audit backend succeeded in logging the request"))
+		}
+	} else {
+		if len(a.backends) > 0 {
+			e, err := audit.NewEvent(audit.RequestType)
+			if err != nil {
+				retErr = multierror.Append(retErr, err)
+			}
+
+			e.Data = in
+
+			start := time.Now()
+			_, err = a.broker.Send(ctx, eventlogger.EventType(event.AuditType.String()), e)
+			// TODO: old behavior includes the name (path) for the audit device,
+			// but we cannot know this anymore, do we just omit it, or include
+			// something like 'all'?
+			// If we can later change the semantics of the eventbroker to report back
+			// as sinks complete then we might be able to reinstate the old behavior.
+			metrics.MeasureSince([]string{"audit", "log_request"}, start)
+			if err != nil {
+				retErr = multierror.Append(retErr, err)
+			}
 		}
-	}
-	if !anyLogged && len(a.backends) > 0 {
-		retErr = multierror.Append(retErr, fmt.Errorf("no audit backend succeeded in logging the request"))
 	}
 
 	return retErr.ErrorOrNil()
@@ -244,34 +272,57 @@ func (a *AuditBroker) LogResponse(ctx context.Context, in *logical.LogInput, hea
 	}()
 
 	// Ensure at least one backend logs
-	anyLogged := false
-	for name, be := range a.backends {
-		in.Request.Headers = nil
-		transHeaders, thErr := headersConfig.ApplyConfig(ctx, headers, be.backend)
-		if thErr != nil {
-			a.logger.Error("backend failed to include headers", "backend", name, "error", thErr)
-			continue
-		}
-		in.Request.Headers = transHeaders
+	if a.broker == nil {
+		anyLogged := false
+		for name, be := range a.backends {
+			in.Request.Headers = nil
+			transHeaders, thErr := headersConfig.ApplyConfig(ctx, headers, be.backend)
+			if thErr != nil {
+				a.logger.Error("backend failed to include headers", "backend", name, "error", thErr)
+				continue
+			}
+			in.Request.Headers = transHeaders
 
-		start := time.Now()
-		lrErr := be.backend.LogResponse(ctx, in)
-		metrics.MeasureSince([]string{"audit", name, "log_response"}, start)
-		if lrErr != nil {
-			a.logger.Error("backend failed to log response", "backend", name, "error", lrErr)
-		} else {
-			anyLogged = true
+			start := time.Now()
+			lrErr := be.backend.LogResponse(ctx, in)
+			metrics.MeasureSince([]string{"audit", name, "log_response"}, start)
+			if lrErr != nil {
+				a.logger.Error("backend failed to log response", "backend", name, "error", lrErr)
+			} else {
+				anyLogged = true
+			}
+		}
+		if !anyLogged && len(a.backends) > 0 {
+			retErr = multierror.Append(retErr, fmt.Errorf("no audit backend succeeded in logging the response"))
+		}
+	} else {
+		if len(a.backends) > 0 {
+			e, err := audit.NewEvent(audit.ResponseType)
+			if err != nil {
+				return multierror.Append(retErr, err)
+			}
+
+			e.Data = in
+
+			start := time.Now()
+			_, err = a.broker.Send(ctx, eventlogger.EventType(event.AuditType.String()), e)
+			// TODO: old behavior includes the name (path) for the audit device,
+			// but we cannot know this anymore, do we just omit it, or include
+			// something like 'all'?
+			// If we can later change the semantics of the eventbroker to report back
+			// as sinks complete then we might be able to reinstate the old behavior.
+			metrics.MeasureSince([]string{"audit", "log_response"}, start)
+			if err != nil {
+				retErr = multierror.Append(retErr, err)
+			}
 		}
-	}
-	if !anyLogged && len(a.backends) > 0 {
-		retErr = multierror.Append(retErr, fmt.Errorf("no audit backend succeeded in logging the response"))
 	}
 
 	return retErr.ErrorOrNil()
 }
 
 func (a *AuditBroker) Invalidate(ctx context.Context, key string) {
-	// For now we ignore the key as this would only apply to salts. We just
+	// For now, we ignore the key as this would only apply to salts. We just
 	// sort of brute force it on each one.
 	a.Lock()
 	defer a.Unlock()

vault/core.go

@@ -2335,6 +2335,10 @@ func (s standardUnsealStrategy) unseal(ctx context.Context, logger log.Logger, c
 		if err := c.loadAudits(ctx); err != nil {
 			return err
 		}
+		if err := c.setupAuditedHeadersConfig(ctx); err != nil {
+			return err
+		}
+
 		if err := c.setupAudits(ctx); err != nil {
 			return err
 		}
@@ -2349,10 +2353,6 @@ func (s standardUnsealStrategy) unseal(ctx context.Context, logger log.Logger, c
 			return err
 		}
 
-		if err := c.setupAuditedHeadersConfig(ctx); err != nil {
-			return err
-		}
-
 		if err := c.setupCensusAgent(); err != nil {
 			c.logger.Error("skipping reporting for nil agent", "error", err)
 		}