[VAULT-22481] Add audit filtering feature (#24558)

* VAULT-22481: Audit filter node (#24465)

* Initial commit on adding filter nodes for audit

* tests for audit filter

* test: longer filter - more conditions

* copywrite headers

* Check interface for the right type

* Add audit filtering feature (#24554)

* Support filter nodes in backend factories and add some tests

* More tests and cleanup

* Attempt to move control of registration for nodes and pipelines to the audit broker (#24505)

* invert control of the pipelines/nodes to the audit broker vs. within each backend

* update noop audit test code to implement the pipeliner interface

* noop mount path has trailing slash

* attempting to make NoopAudit more friendly

* NoopAudit uses known salt

* Refactor audit.ProcessManual to support filter nodes

* HasFiltering

* rename the pipeliner

* use exported AuditEvent in Filter

* Add tests for registering and deregistering backends on the audit broker

* Add missing licence header to one file, fix a typo in two tests

---------

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

* Add changelog file

* update bexpr datum to use a strong type

* go docs updates

* test path

* PR review comments

* handle scenarios/outcomes from broker.send

* don't need to re-check the complete sinks

* add extra check to deregister to ensure that re-registering non-filtered device sets sink threshold

* Ensure that the multierror is appended before attempting to return it

---------

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

Author: kubawi

audit/entry_filter.go

@@ -0,0 +1,91 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package audit
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/hashicorp/eventlogger"
+	"github.com/hashicorp/go-bexpr"
+	"github.com/hashicorp/vault/helper/namespace"
+	"github.com/hashicorp/vault/internal/observability/event"
+)
+
+var _ eventlogger.Node = (*EntryFilter)(nil)
+
+// NewEntryFilter should be used to create an EntryFilter node.
+// The filter supplied should be in bexpr format and reference fields from logical.LogInputBexpr.
+func NewEntryFilter(filter string) (*EntryFilter, error) {
+	const op = "audit.NewEntryFilter"
+
+	filter = strings.TrimSpace(filter)
+	if filter == "" {
+		return nil, fmt.Errorf("%s: cannot create new audit filter with empty filter expression: %w", op, event.ErrInvalidParameter)
+	}
+
+	eval, err := bexpr.CreateEvaluator(filter)
+	if err != nil {
+		return nil, fmt.Errorf("%s: cannot create new audit filter: %w", op, err)
+	}
+
+	return &EntryFilter{evaluator: eval}, nil
+}
+
+// Reopen is a no-op for the filter node.
+func (*EntryFilter) Reopen() error {
+	return nil
+}
+
+// Type describes the type of this node (filter).
+func (*EntryFilter) Type() eventlogger.NodeType {
+	return eventlogger.NodeTypeFilter
+}
+
+// Process will attempt to parse the incoming event data and decide whether it
+// should be filtered or remain in the pipeline and passed to the next node.
+func (f *EntryFilter) Process(ctx context.Context, e *eventlogger.Event) (*eventlogger.Event, error) {
+	const op = "audit.(EntryFilter).Process"
+
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	default:
+	}
+
+	if e == nil {
+		return nil, fmt.Errorf("%s: event is nil: %w", op, event.ErrInvalidParameter)
+	}
+
+	a, ok := e.Payload.(*AuditEvent)
+	if !ok {
+		return nil, fmt.Errorf("%s: cannot parse event payload: %w", op, event.ErrInvalidParameter)
+	}
+
+	// If we don't have data to process, then we're done.
+	if a.Data == nil {
+		return nil, nil
+	}
+
+	ns, err := namespace.FromContext(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("%s: cannot obtain namespace: %w", op, err)
+	}
+
+	datum := a.Data.BexprDatum(ns.Path)
+
+	result, err := f.evaluator.Evaluate(datum)
+	if err != nil {
+		return nil, fmt.Errorf("%s: unable to evaluate filter: %w", op, err)
+	}
+
+	if result {
+		// Allow this event to carry on through the pipeline.
+		return e, nil
+	}
+
+	// End process of this pipeline.
+	return nil, nil
+}

audit/entry_filter_test.go

@@ -0,0 +1,249 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package audit
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/eventlogger"
+	"github.com/hashicorp/vault/helper/namespace"
+	"github.com/hashicorp/vault/internal/observability/event"
+	"github.com/hashicorp/vault/sdk/logical"
+	"github.com/stretchr/testify/require"
+)
+
+// TestEntryFilter_NewEntryFilter tests that we can create EntryFilter types correctly.
+func TestEntryFilter_NewEntryFilter(t *testing.T) {
+	t.Parallel()
+
+	tests := map[string]struct {
+		Filter               string
+		IsErrorExpected      bool
+		ExpectedErrorMessage string
+	}{
+		"empty-filter": {
+			Filter:               "",
+			IsErrorExpected:      true,
+			ExpectedErrorMessage: "audit.NewEntryFilter: cannot create new audit filter with empty filter expression: invalid parameter",
+		},
+		"spacey-filter": {
+			Filter:               "    ",
+			IsErrorExpected:      true,
+			ExpectedErrorMessage: "audit.NewEntryFilter: cannot create new audit filter with empty filter expression: invalid parameter",
+		},
+		"bad-filter": {
+			Filter:               "____",
+			IsErrorExpected:      true,
+			ExpectedErrorMessage: "audit.NewEntryFilter: cannot create new audit filter",
+		},
+		"good-filter": {
+			Filter:          "foo == bar",
+			IsErrorExpected: false,
+		},
+	}
+
+	for name, tc := range tests {
+		name := name
+		tc := tc
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			f, err := NewEntryFilter(tc.Filter)
+			switch {
+			case tc.IsErrorExpected:
+				require.ErrorContains(t, err, tc.ExpectedErrorMessage)
+				require.Nil(t, f)
+			default:
+				require.NoError(t, err)
+				require.NotNil(t, f)
+			}
+		})
+	}
+}
+
+// TestEntryFilter_Reopen ensures we can reopen the filter node.
+func TestEntryFilter_Reopen(t *testing.T) {
+	t.Parallel()
+
+	f := &EntryFilter{}
+	res := f.Reopen()
+	require.Nil(t, res)
+}
+
+// TestEntryFilter_Type ensures we always return the right type for this node.
+func TestEntryFilter_Type(t *testing.T) {
+	t.Parallel()
+
+	f := &EntryFilter{}
+	require.Equal(t, eventlogger.NodeTypeFilter, f.Type())
+}
+
+// TestEntryFilter_Process_ContextDone ensures that we stop processing the event
+// if the context was cancelled.
+func TestEntryFilter_Process_ContextDone(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	// Explicitly cancel the context
+	cancel()
+
+	l, err := NewEntryFilter("foo == bar")
+	require.NoError(t, err)
+
+	// Fake audit event
+	a, err := NewEvent(RequestType)
+	require.NoError(t, err)
+
+	// Fake event logger event
+	e := &eventlogger.Event{
+		Type:      eventlogger.EventType(event.AuditType.String()),
+		CreatedAt: time.Now(),
+		Formatted: make(map[string][]byte),
+		Payload:   a,
+	}
+
+	e2, err := l.Process(ctx, e)
+
+	require.Error(t, err)
+	require.ErrorContains(t, err, "context canceled")
+
+	// Ensure that the pipeline won't continue.
+	require.Nil(t, e2)
+}
+
+// TestEntryFilter_Process_NilEvent ensures we receive the right error when the
+// event we are trying to process is nil.
+func TestEntryFilter_Process_NilEvent(t *testing.T) {
+	t.Parallel()
+
+	l, err := NewEntryFilter("foo == bar")
+	require.NoError(t, err)
+	e, err := l.Process(context.Background(), nil)
+	require.Error(t, err)
+	require.EqualError(t, err, "audit.(EntryFilter).Process: event is nil: invalid parameter")
+
+	// Ensure that the pipeline won't continue.
+	require.Nil(t, e)
+}
+
+// TestEntryFilter_Process_BadPayload ensures we receive the correct error when
+// attempting to process an event with a payload that cannot be parsed back to
+// an audit event.
+func TestEntryFilter_Process_BadPayload(t *testing.T) {
+	t.Parallel()
+
+	l, err := NewEntryFilter("foo == bar")
+	require.NoError(t, err)
+
+	e := &eventlogger.Event{
+		Type:      eventlogger.EventType(event.AuditType.String()),
+		CreatedAt: time.Now(),
+		Formatted: make(map[string][]byte),
+		Payload:   nil,
+	}
+
+	e2, err := l.Process(context.Background(), e)
+	require.Error(t, err)
+	require.EqualError(t, err, "audit.(EntryFilter).Process: cannot parse event payload: invalid parameter")
+
+	// Ensure that the pipeline won't continue.
+	require.Nil(t, e2)
+}
+
+// TestEntryFilter_Process_NoAuditDataInPayload ensure we stop processing a pipeline
+// when the data in the audit event is nil.
+func TestEntryFilter_Process_NoAuditDataInPayload(t *testing.T) {
+	t.Parallel()
+
+	l, err := NewEntryFilter("foo == bar")
+	require.NoError(t, err)
+
+	a, err := NewEvent(RequestType)
+	require.NoError(t, err)
+
+	// Ensure audit data is nil
+	a.Data = nil
+
+	e := &eventlogger.Event{
+		Type:      eventlogger.EventType(event.AuditType.String()),
+		CreatedAt: time.Now(),
+		Formatted: make(map[string][]byte),
+		Payload:   a,
+	}
+
+	e2, err := l.Process(context.Background(), e)
+
+	// Make sure we get the 'nil, nil' response to stop processing this pipeline.
+	require.NoError(t, err)
+	require.Nil(t, e2)
+}
+
+// TestEntryFilter_Process_FilterSuccess tests that when a filter matches we
+// receive no error and the event is not nil so it continues in the pipeline.
+func TestEntryFilter_Process_FilterSuccess(t *testing.T) {
+	t.Parallel()
+
+	l, err := NewEntryFilter("mount_type == juan")
+	require.NoError(t, err)
+
+	a, err := NewEvent(RequestType)
+	require.NoError(t, err)
+
+	a.Data = &logical.LogInput{
+		Request: &logical.Request{
+			Operation: logical.CreateOperation,
+			MountType: "juan",
+		},
+	}
+
+	e := &eventlogger.Event{
+		Type:      eventlogger.EventType(event.AuditType.String()),
+		CreatedAt: time.Now(),
+		Formatted: make(map[string][]byte),
+		Payload:   a,
+	}
+
+	ctx := namespace.ContextWithNamespace(context.Background(), namespace.RootNamespace)
+
+	e2, err := l.Process(ctx, e)
+
+	require.NoError(t, err)
+	require.NotNil(t, e2)
+}
+
+// TestEntryFilter_Process_FilterFail tests that when a filter fails to match we
+// receive no error, but also the event is nil so that the pipeline completes.
+func TestEntryFilter_Process_FilterFail(t *testing.T) {
+	t.Parallel()
+
+	l, err := NewEntryFilter("mount_type == john and operation == create and namespace == root")
+	require.NoError(t, err)
+
+	a, err := NewEvent(RequestType)
+	require.NoError(t, err)
+
+	a.Data = &logical.LogInput{
+		Request: &logical.Request{
+			Operation: logical.CreateOperation,
+			MountType: "juan",
+		},
+	}
+
+	e := &eventlogger.Event{
+		Type:      eventlogger.EventType(event.AuditType.String()),
+		CreatedAt: time.Now(),
+		Formatted: make(map[string][]byte),
+		Payload:   a,
+	}
+
+	ctx := namespace.ContextWithNamespace(context.Background(), namespace.RootNamespace)
+
+	e2, err := l.Process(ctx, e)
+
+	require.NoError(t, err)
+	require.Nil(t, e2)
+}

audit/entry_formatter.go

@@ -11,16 +11,13 @@ import (
 	"strings"
 	"time"
 
-	"github.com/jefferai/jsonx"
-
-	"github.com/hashicorp/vault/helper/namespace"
-	"github.com/hashicorp/vault/sdk/logical"
-
 	"github.com/go-jose/go-jose/v3/jwt"
+	"github.com/hashicorp/eventlogger"
+	"github.com/hashicorp/vault/helper/namespace"
 	"github.com/hashicorp/vault/internal/observability/event"
 	"github.com/hashicorp/vault/sdk/helper/jsonutil"
-
-	"github.com/hashicorp/eventlogger"
+	"github.com/hashicorp/vault/sdk/logical"
+	"github.com/jefferai/jsonx"
 )
 
 var (
@@ -29,7 +26,7 @@ var (
 )
 
 // NewEntryFormatter should be used to create an EntryFormatter.
-// Accepted options: WithPrefix, WithHeaderFormatter.
+// Accepted options: WithHeaderFormatter, WithPrefix.
 func NewEntryFormatter(config FormatterConfig, salter Salter, opt ...Option) (*EntryFormatter, error) {
 	const op = "audit.NewEntryFormatter"