[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>

builtin/audit/socket/backend.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"net"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -21,83 +22,76 @@ import (
 	"github.com/hashicorp/vault/sdk/logical"
 )
 
-func Factory(ctx context.Context, conf *audit.BackendConfig, useEventLogger bool, headersConfig audit.HeaderFormatter) (audit.Backend, error) {
+var _ audit.Backend = (*Backend)(nil)
+
+// Backend is the audit backend for the socket audit transport.
+type Backend struct {
+	sync.Mutex
+	address       string
+	connection    net.Conn
+	formatter     *audit.EntryFormatterWriter
+	formatConfig  audit.FormatterConfig
+	name          string
+	nodeIDList    []eventlogger.NodeID
+	nodeMap       map[eventlogger.NodeID]eventlogger.Node
+	salt          *salt.Salt
+	saltConfig    *salt.Config
+	saltMutex     sync.RWMutex
+	saltView      logical.Storage
+	socketType    string
+	writeDuration time.Duration
+}
+
+func Factory(_ context.Context, conf *audit.BackendConfig, useEventLogger bool, headersConfig audit.HeaderFormatter) (audit.Backend, error) {
+	const op = "socket.Factory"
+
 	if conf.SaltConfig == nil {
-		return nil, fmt.Errorf("nil salt config")
+		return nil, fmt.Errorf("%s: nil salt config", op)
 	}
+
 	if conf.SaltView == nil {
-		return nil, fmt.Errorf("nil salt view")
+		return nil, fmt.Errorf("%s: nil salt view", op)
 	}
 
 	address, ok := conf.Config["address"]
 	if !ok {
-		return nil, fmt.Errorf("address is required")
+		return nil, fmt.Errorf("%s: address is required", op)
 	}
 
 	socketType, ok := conf.Config["socket_type"]
 	if !ok {
 		socketType = "tcp"
 	}
+
 	writeDeadline, ok := conf.Config["write_timeout"]
 	if !ok {
 		writeDeadline = "2s"
 	}
+
 	writeDuration, err := parseutil.ParseDurationSecond(writeDeadline)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("%s: failed to parse 'write_timeout': %w", op, err)
 	}
 
-	var cfgOpts []audit.Option
-
-	if format, ok := conf.Config["format"]; ok {
-		cfgOpts = append(cfgOpts, audit.WithFormat(format))
-	}
-
-	// Check if hashing of accessor is disabled
-	if hmacAccessorRaw, ok := conf.Config["hmac_accessor"]; ok {
-		v, err := strconv.ParseBool(hmacAccessorRaw)
-		if err != nil {
-			return nil, err
-		}
-		cfgOpts = append(cfgOpts, audit.WithHMACAccessor(v))
-	}
-
-	// Check if raw logging is enabled
-	if raw, ok := conf.Config["log_raw"]; ok {
-		v, err := strconv.ParseBool(raw)
-		if err != nil {
-			return nil, err
-		}
-		cfgOpts = append(cfgOpts, audit.WithRaw(v))
-	}
-
-	if elideListResponsesRaw, ok := conf.Config["elide_list_responses"]; ok {
-		v, err := strconv.ParseBool(elideListResponsesRaw)
-		if err != nil {
-			return nil, err
-		}
-		cfgOpts = append(cfgOpts, audit.WithElision(v))
-	}
-
-	cfg, err := audit.NewFormatterConfig(cfgOpts...)
+	cfg, err := formatterConfig(conf.Config)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("%s: failed to create formatter config: %w", op, err)
 	}
 
 	b := &Backend{
-		saltConfig:   conf.SaltConfig,
-		saltView:     conf.SaltView,
-		formatConfig: cfg,
-
-		writeDuration: writeDuration,
 		address:       address,
+		formatConfig:  cfg,
+		name:          conf.MountPath,
+		saltConfig:    conf.SaltConfig,
+		saltView:      conf.SaltView,
 		socketType:    socketType,
+		writeDuration: writeDuration,
 	}
 
 	// Configure the formatter for either case.
-	f, err := audit.NewEntryFormatter(b.formatConfig, b, audit.WithHeaderFormatter(headersConfig))
+	f, err := audit.NewEntryFormatter(cfg, b, audit.WithHeaderFormatter(headersConfig))
 	if err != nil {
-		return nil, fmt.Errorf("error creating formatter: %w", err)
+		return nil, fmt.Errorf("%s: error creating formatter: %w", op, err)
 	}
 	var w audit.Writer
 	switch b.formatConfig.RequiredFormat {
@@ -109,72 +103,44 @@ func Factory(ctx context.Context, conf *audit.BackendConfig, useEventLogger bool
 
 	fw, err := audit.NewEntryFormatterWriter(b.formatConfig, f, w)
 	if err != nil {
-		return nil, fmt.Errorf("error creating formatter writer: %w", err)
+		return nil, fmt.Errorf("%s: error creating formatter writer: %w", op, err)
 	}
 
 	b.formatter = fw
 
 	if useEventLogger {
-		var opts []event.Option
-
-		if socketType, ok := conf.Config["socket_type"]; ok {
-			opts = append(opts, event.WithSocketType(socketType))
-		}
-
-		if writeDeadline, ok := conf.Config["write_timeout"]; ok {
-			opts = append(opts, event.WithMaxDuration(writeDeadline))
-		}
-
-		b.nodeIDList = make([]eventlogger.NodeID, 2)
+		b.nodeIDList = []eventlogger.NodeID{}
 		b.nodeMap = make(map[eventlogger.NodeID]eventlogger.Node)
 
-		formatterNodeID, err := event.GenerateNodeID()
+		err := b.configureFilterNode(conf.Config["filter"])
 		if err != nil {
-			return nil, fmt.Errorf("error generating random NodeID for formatter node: %w", err)
+			return nil, fmt.Errorf("%s: error configuring filter node: %w", op, err)
 		}
-		b.nodeIDList[0] = formatterNodeID
-		b.nodeMap[formatterNodeID] = f
 
-		n, err := event.NewSocketSink(b.formatConfig.RequiredFormat.String(), address, opts...)
-		if err != nil {
-			return nil, fmt.Errorf("error creating socket sink node: %w", err)
+		opts := []audit.Option{
+			audit.WithHeaderFormatter(headersConfig),
 		}
-		sinkNode := &audit.SinkWrapper{Name: conf.MountPath, Sink: n}
-		sinkNodeID, err := event.GenerateNodeID()
+
+		err = b.configureFormatterNode(cfg, opts...)
 		if err != nil {
-			return nil, fmt.Errorf("error generating random NodeID for sink node: %w", err)
+			return nil, fmt.Errorf("%s: error configuring formatter node: %w", op, err)
+		}
+
+		sinkOpts := []event.Option{
+			event.WithSocketType(socketType),
+			event.WithMaxDuration(writeDeadline),
+		}
+
+		err = b.configureSinkNode(conf.MountPath, address, cfg.RequiredFormat.String(), sinkOpts...)
+		if err != nil {
+			return nil, fmt.Errorf("%s: error configuring sink node: %w", op, err)
 		}
-		b.nodeIDList[1] = sinkNodeID
-		b.nodeMap[sinkNodeID] = sinkNode
 	}
 
 	return b, nil
 }
 
-// Backend is the audit backend for the socket audit transport.
-type Backend struct {
-	connection net.Conn
-
-	formatter    *audit.EntryFormatterWriter
-	formatConfig audit.FormatterConfig
-
-	writeDuration time.Duration
-	address       string
-	socketType    string
-
-	sync.Mutex
-
-	saltMutex  sync.RWMutex
-	salt       *salt.Salt
-	saltConfig *salt.Config
-	saltView   logical.Storage
-
-	nodeIDList []eventlogger.NodeID
-	nodeMap    map[eventlogger.NodeID]eventlogger.Node
-}
-
-var _ audit.Backend = (*Backend)(nil)
-
+// Deprecated: Use eventlogger.
 func (b *Backend) LogRequest(ctx context.Context, in *logical.LogInput) error {
 	var buf bytes.Buffer
 	if err := b.formatter.FormatAndWriteRequest(ctx, &buf, in); err != nil {
@@ -198,6 +164,7 @@ func (b *Backend) LogRequest(ctx context.Context, in *logical.LogInput) error {
 	return err
 }
 
+// Deprecated: Use eventlogger.
 func (b *Backend) LogResponse(ctx context.Context, in *logical.LogInput) error {
 	var buf bytes.Buffer
 	if err := b.formatter.FormatAndWriteResponse(ctx, &buf, in); err != nil {
@@ -256,6 +223,7 @@ func (b *Backend) LogTestMessage(ctx context.Context, in *logical.LogInput, conf
 	return err
 }
 
+// Deprecated: Use eventlogger.
 func (b *Backend) write(ctx context.Context, buf []byte) error {
 	if b.connection == nil {
 		if err := b.reconnect(ctx); err != nil {
@@ -276,6 +244,7 @@ func (b *Backend) write(ctx context.Context, buf []byte) error {
 	return nil
 }
 
+// Deprecated: Use eventlogger.
 func (b *Backend) reconnect(ctx context.Context) error {
 	if b.connection != nil {
 		b.connection.Close()
@@ -317,12 +286,12 @@ func (b *Backend) Salt(ctx context.Context) (*salt.Salt, error) {
 	if b.salt != nil {
 		return b.salt, nil
 	}
-	salt, err := salt.NewSalt(ctx, b.saltView, b.saltConfig)
+	s, err := salt.NewSalt(ctx, b.saltView, b.saltConfig)
 	if err != nil {
 		return nil, err
 	}
-	b.salt = salt
-	return salt, nil
+	b.salt = s
+	return s, nil
 }
 
 func (b *Backend) Invalidate(_ context.Context) {
@@ -331,20 +300,146 @@ func (b *Backend) Invalidate(_ context.Context) {
 	b.salt = nil
 }
 
-// RegisterNodesAndPipeline registers the nodes and a pipeline as required by
-// the audit.Backend interface.
-func (b *Backend) RegisterNodesAndPipeline(broker *eventlogger.Broker, name string) error {
-	for id, node := range b.nodeMap {
-		if err := broker.RegisterNode(id, node, eventlogger.WithNodeRegistrationPolicy(eventlogger.DenyOverwrite)); err != nil {
-			return err
+// formatterConfig creates the configuration required by a formatter node using
+// the config map supplied to the factory.
+func formatterConfig(config map[string]string) (audit.FormatterConfig, error) {
+	const op = "socket.formatterConfig"
+
+	var cfgOpts []audit.Option
+
+	if format, ok := config["format"]; ok {
+		cfgOpts = append(cfgOpts, audit.WithFormat(format))
+	}
+
+	// Check if hashing of accessor is disabled
+	if hmacAccessorRaw, ok := config["hmac_accessor"]; ok {
+		v, err := strconv.ParseBool(hmacAccessorRaw)
+		if err != nil {
+			return audit.FormatterConfig{}, fmt.Errorf("%s: unable to parse 'hmac_accessor': %w", op, err)
 		}
+		cfgOpts = append(cfgOpts, audit.WithHMACAccessor(v))
 	}
 
-	pipeline := eventlogger.Pipeline{
-		PipelineID: eventlogger.PipelineID(name),
-		EventType:  eventlogger.EventType(event.AuditType.String()),
-		NodeIDs:    b.nodeIDList,
+	// Check if raw logging is enabled
+	if raw, ok := config["log_raw"]; ok {
+		v, err := strconv.ParseBool(raw)
+		if err != nil {
+			return audit.FormatterConfig{}, fmt.Errorf("%s: unable to parse 'log_raw': %w", op, err)
+		}
+		cfgOpts = append(cfgOpts, audit.WithRaw(v))
 	}
 
-	return broker.RegisterPipeline(pipeline, eventlogger.WithPipelineRegistrationPolicy(eventlogger.DenyOverwrite))
+	if elideListResponsesRaw, ok := config["elide_list_responses"]; ok {
+		v, err := strconv.ParseBool(elideListResponsesRaw)
+		if err != nil {
+			return audit.FormatterConfig{}, fmt.Errorf("%s: unable to parse 'elide_list_responses': %w", op, err)
+		}
+		cfgOpts = append(cfgOpts, audit.WithElision(v))
+	}
+
+	return audit.NewFormatterConfig(cfgOpts...)
+}
+
+// configureFilterNode is used to configure a filter node and associated ID on the Backend.
+func (b *Backend) configureFilterNode(filter string) error {
+	const op = "socket.(Backend).configureFilterNode"
+
+	filter = strings.TrimSpace(filter)
+	if filter == "" {
+		return nil
+	}
+
+	filterNodeID, err := event.GenerateNodeID()
+	if err != nil {
+		return fmt.Errorf("%s: error generating random NodeID for filter node: %w", op, err)
+	}
+
+	filterNode, err := audit.NewEntryFilter(filter)
+	if err != nil {
+		return fmt.Errorf("%s: error creating filter node: %w", op, err)
+	}
+
+	b.nodeIDList = append(b.nodeIDList, filterNodeID)
+	b.nodeMap[filterNodeID] = filterNode
+	return nil
+}
+
+// configureFormatterNode is used to configure a formatter node and associated ID on the Backend.
+func (b *Backend) configureFormatterNode(formatConfig audit.FormatterConfig, opts ...audit.Option) error {
+	const op = "socket.(Backend).configureFormatterNode"
+
+	formatterNodeID, err := event.GenerateNodeID()
+	if err != nil {
+		return fmt.Errorf("%s: error generating random NodeID for formatter node: %w", op, err)
+	}
+
+	formatterNode, err := audit.NewEntryFormatter(formatConfig, b, opts...)
+	if err != nil {
+		return fmt.Errorf("%s: error creating formatter: %w", op, err)
+	}
+
+	b.nodeIDList = append(b.nodeIDList, formatterNodeID)
+	b.nodeMap[formatterNodeID] = formatterNode
+	return nil
+}
+
+// configureSinkNode is used to configure a sink node and associated ID on the Backend.
+func (b *Backend) configureSinkNode(name string, address string, format string, opts ...event.Option) error {
+	const op = "socket.(Backend).configureSinkNode"
+
+	name = strings.TrimSpace(name)
+	if name == "" {
+		return fmt.Errorf("%s: name is required: %w", op, event.ErrInvalidParameter)
+	}
+
+	address = strings.TrimSpace(address)
+	if address == "" {
+		return fmt.Errorf("%s: address is required: %w", op, event.ErrInvalidParameter)
+	}
+
+	format = strings.TrimSpace(format)
+	if format == "" {
+		return fmt.Errorf("%s: format is required: %w", op, event.ErrInvalidParameter)
+	}
+
+	sinkNodeID, err := event.GenerateNodeID()
+	if err != nil {
+		return fmt.Errorf("%s: error generating random NodeID for sink node: %w", op, err)
+	}
+
+	n, err := event.NewSocketSink(address, format, opts...)
+	if err != nil {
+		return fmt.Errorf("%s: error creating socket sink node: %w", op, err)
+	}
+
+	sinkNode := &audit.SinkWrapper{Name: name, Sink: n}
+
+	b.nodeIDList = append(b.nodeIDList, sinkNodeID)
+	b.nodeMap[sinkNodeID] = sinkNode
+	return nil
+}
+
+// Name for this backend, this would ideally correspond to the mount path for the audit device.
+func (b *Backend) Name() string {
+	return b.name
+}
+
+// Nodes returns the nodes which should be used by the event framework to process audit entries.
+func (b *Backend) Nodes() map[eventlogger.NodeID]eventlogger.Node {
+	return b.nodeMap
+}
+
+// NodeIDs returns the IDs of the nodes, in the order they are required.
+func (b *Backend) NodeIDs() []eventlogger.NodeID {
+	return b.nodeIDList
+}
+
+// EventType returns the event type for the backend.
+func (b *Backend) EventType() eventlogger.EventType {
+	return eventlogger.EventType(event.AuditType.String())
+}
+
+// HasFiltering determines if the first node for the pipeline is an eventlogger.NodeTypeFilter.
+func (b *Backend) HasFiltering() bool {
+	return len(b.nodeIDList) > 0 && b.nodeMap[b.nodeIDList[0]].Type() == eventlogger.NodeTypeFilter
 }