scottk/vault
1
0
Fork 0
mirror of https://github.com/optim-enterprises-bv/vault.git synced 2025-10-29 09:42:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

VAULT-18284: Audit refactor packages (#21972)

Browse Source 
* initial git mv to rename 'audit' packages

* remove 'Audit' prefix from structs inside audit package

* refactor of event/audit pacakges

* EventFormatter => EntryFormatter

* 'AuditFormat' => EntryFormat

* Use NewFormatterConfig func

---------

Co-authored-by: Marc Boudreau <marc.boudreau@hashicorp.com>
This commit is contained in:
Peter Wilson
2023-07-20 18:32:06 +01:00
committed by GitHub
parent a4f67a6b2b
commit 31074bc448
35 changed files with 3236 additions and 2798 deletions
Download Patch File Download Diff File Expand all files Collapse all files

250
internal/observability/event/audit_sink_file.go
View File

@@ -1,250 +0,0 @@
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: MPL-2.0
package event
import (
"bytes"
"context"
"fmt"
"io"
"os"
"path/filepath"
"strings"
"sync"
"github.com/hashicorp/eventlogger"
)
// defaultFileMode is the default file permissions (read/write for everyone).
const (
defaultFileMode = 0o600
discard = "discard"
stdout = "stdout"
)
// AuditFileSink is a sink node which handles writing audit events to file.
type AuditFileSink struct {
file *os.File
fileLock sync.RWMutex
fileMode os.FileMode
path string
format auditFormat
prefix string
}
// NewAuditFileSink should be used to create a new AuditFileSink.
// Accepted options: WithFileMode and WithPrefix.
func NewAuditFileSink(path string, format auditFormat, opt ...Option) (*AuditFileSink, error) {
const op = "event.NewAuditFileSink"
// Parse and check path
p := strings.TrimSpace(path)
switch {
case p == "":
return nil, fmt.Errorf("%s: path is required", op)
case strings.EqualFold(path, stdout):
p = stdout
case strings.EqualFold(path, discard):
p = discard
}
// Validate format
if err := format.validate(); err != nil {
return nil, fmt.Errorf("%s: invalid format: %w", op, err)
}
opts, err := getOpts(opt...)
if err != nil {
return nil, fmt.Errorf("%s: error applying options: %w", op, err)
}
mode := os.FileMode(defaultFileMode)
// If we got an optional file mode supplied and our path isn't a special keyword
// then we should use the supplied file mode, or maintain the existing file mode.
if opts.withFileMode != nil {
switch {
case p == stdout:
case p == discard:
case *opts.withFileMode == 0: // Maintain the existing file's mode when set to "0000".
fileInfo, err := os.Stat(path)
if err != nil {
return nil, fmt.Errorf("%s: unable to determine existing file mode: %w", op, err)
}
mode = fileInfo.Mode()
default:
mode = *opts.withFileMode
}
}
return &AuditFileSink{
file: nil,
fileLock: sync.RWMutex{},
fileMode: mode,
format: format,
path: p,
prefix: opts.withPrefix,
}, nil
}
// Process handles writing the event to the file sink.
func (f *AuditFileSink) Process(ctx context.Context, e *eventlogger.Event) (*eventlogger.Event, error) {
const op = "event.(AuditFileSink).Process"
select {
case <-ctx.Done():
return nil, ctx.Err()
default:
}
if e == nil {
return nil, fmt.Errorf("%s: event is nil: %w", op, ErrInvalidParameter)
}
// 'discard' path means we just do nothing and pretend we're done.
if f.path == discard {
return nil, nil
}
formatted, found := e.Format(f.format.String())
if !found {
return nil, fmt.Errorf("%s: unable to retrieve event formatted as %q", op, f.format)
}
err := f.log(formatted)
if err != nil {
return nil, fmt.Errorf("%s: error writing file for audit sink: %w", op, err)
}
// return nil for the event to indicate the pipeline is complete.
return nil, nil
}
// Reopen handles closing and reopening the file.
func (f *AuditFileSink) Reopen() error {
const op = "event.(AuditFileSink).Reopen"
switch f.path {
case stdout, discard:
return nil
}
f.fileLock.Lock()
defer f.fileLock.Unlock()
if f.file == nil {
return f.open()
}
err := f.file.Close()
// Set to nil here so that even if we error out, on the next access open() will be tried.
f.file = nil
if err != nil {
return fmt.Errorf("%s: unable to close file for re-opening on audit sink: %w", op, err)
}
return f.open()
}
// Type describes the type of this node (sink).
func (f *AuditFileSink) Type() eventlogger.NodeType {
return eventlogger.NodeTypeSink
}
// open attempts to open a file at the sink's path, with the sink's fileMode permissions
// if one is not already open.
// It doesn't have any locking and relies on calling functions of AuditFileSink to
// handle this (e.g. log and Reopen methods).
func (f *AuditFileSink) open() error {
const op = "event.(AuditFileSink).open"
if f.file != nil {
return nil
}
if err := os.MkdirAll(filepath.Dir(f.path), f.fileMode); err != nil {
return fmt.Errorf("%s: unable to create file %q: %w", op, f.path, err)
}
var err error
f.file, err = os.OpenFile(f.path, os.O_APPEND|os.O_WRONLY|os.O_CREATE, f.fileMode)
if err != nil {
return fmt.Errorf("%s: unable to open file for audit sink: %w", op, err)
}
// Change the file mode in case the log file already existed.
// We special case '/dev/null' since we can't chmod it, and bypass if the mode is zero.
switch f.path {
case "/dev/null":
default:
if f.fileMode != 0 {
err = os.Chmod(f.path, f.fileMode)
if err != nil {
return fmt.Errorf("%s: unable to change file %q permissions '%v' for audit sink: %w", op, f.path, f.fileMode, err)
}
}
}
return nil
}
// log writes the buffer to the file.
// It acquires a lock on the file to do this.
func (f *AuditFileSink) log(data []byte) error {
const op = "event.(AuditFileSink).log"
f.fileLock.Lock()
defer f.fileLock.Unlock()
reader := bytes.NewReader(data)
var writer io.Writer
switch {
case f.path == stdout:
writer = os.Stdout
default:
if err := f.open(); err != nil {
return fmt.Errorf("%s: unable to open file for audit sink: %w", op, err)
}
writer = f.file
}
// Write prefix before the data if required.
if f.prefix != "" {
_, err := writer.Write([]byte(f.prefix))
if err != nil {
return fmt.Errorf("%s: unable to write prefix %q for audit sink: %w", op, f.prefix, err)
}
}
if _, err := reader.WriteTo(writer); err == nil {
return nil
} else if f.path == stdout {
// If writing to stdout there's no real reason to think anything would change on retry.
return fmt.Errorf("%s: unable write to %q: %w", op, f.path, err)
}
// Otherwise, opportunistically try to re-open the FD, once per call (1 retry attempt).
err := f.file.Close()
if err != nil {
return fmt.Errorf("%s: unable to close file for audit sink: %w", op, err)
}
f.file = nil
if err := f.open(); err != nil {
return fmt.Errorf("%s: unable to re-open file for audit sink: %w", op, err)
}
_, err = reader.Seek(0, io.SeekStart)
if err != nil {
return fmt.Errorf("%s: unable to seek to start of file for audit sink: %w", op, err)
}
_, err = reader.WriteTo(writer)
if err != nil {
return fmt.Errorf("%s: unable to re-write to file for audit sink: %w", op, err)
}
return nil
}

415
internal/observability/event/audit_sink_file_test.go
View File

@@ -1,415 +0,0 @@
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: MPL-2.0
package event
import (
"os"
"path/filepath"
"testing"
"time"
"github.com/hashicorp/vault/sdk/logical"
vaultaudit "github.com/hashicorp/vault/audit"
"github.com/hashicorp/vault/helper/namespace"
"github.com/hashicorp/eventlogger"
"github.com/stretchr/testify/require"
)
// TestAuditFileSink_Type ensures that the node is a 'sink' type.
func TestAuditFileSink_Type(t *testing.T) {
f, err := NewAuditFileSink(t.TempDir(), AuditFormatJSON)
require.NoError(t, err)
require.NotNil(t, f)
require.Equal(t, eventlogger.NodeTypeSink, f.Type())
}
// TestNewAuditFileSink tests creation of an AuditFileSink.
func TestNewAuditFileSink(t *testing.T) {
tests := map[string]struct {
IsTempDirPath bool // Path should contain the filename if temp dir is true
Path string
Format auditFormat
Options []Option
IsErrorExpected bool
ExpectedErrorMessage string
// Expected values of AuditFileSink
ExpectedFileMode os.FileMode
ExpectedFormat auditFormat
ExpectedPath string
ExpectedPrefix string
}{
"default-values": {
IsErrorExpected: true,
ExpectedErrorMessage: "event.NewAuditFileSink: path is required",
},
"spacey-path": {
Path: " ",
Format: AuditFormatJSON,
IsErrorExpected: true,
ExpectedErrorMessage: "event.NewAuditFileSink: path is required",
},
"bad-format": {
Path: "qwerty",
Format: "squirrels",
IsErrorExpected: true,
ExpectedErrorMessage: "event.NewAuditFileSink: invalid format: event.(auditFormat).validate: 'squirrels' is not a valid format: invalid parameter",
},
"path-not-exist-valid-format-file-mode": {
Path: "qwerty",
Format: AuditFormatJSON,
Options: []Option{WithFileMode("00755")},
IsErrorExpected: false,
ExpectedPath: "qwerty",
ExpectedFormat: AuditFormatJSON,
ExpectedPrefix: "",
ExpectedFileMode: os.FileMode(0o755),
},
"valid-path-no-format": {
IsTempDirPath: true,
Path: "vault.log",
IsErrorExpected: true,
ExpectedErrorMessage: "event.NewAuditFileSink: invalid format: event.(auditFormat).validate: '' is not a valid format: invalid parameter",
},
"valid-path-and-format": {
IsTempDirPath: true,
Path: "vault.log",
Format: AuditFormatJSON,
IsErrorExpected: false,
ExpectedFileMode: defaultFileMode,
ExpectedFormat: AuditFormatJSON,
ExpectedPrefix: "",
},
"file-mode-not-default-or-zero": {
Path: "vault.log",
Format: AuditFormatJSON,
Options: []Option{WithFileMode("0007")},
IsTempDirPath: true,
IsErrorExpected: false,
ExpectedFormat: AuditFormatJSON,
ExpectedPrefix: "",
ExpectedFileMode: 0o007,
},
"path-stdout": {
Path: "stdout",
Format: AuditFormatJSON,
Options: []Option{WithFileMode("0007")}, // Will be ignored as stdout
IsTempDirPath: false,
IsErrorExpected: false,
ExpectedPath: "stdout",
ExpectedFormat: AuditFormatJSON,
ExpectedPrefix: "",
ExpectedFileMode: defaultFileMode,
},
"path-discard": {
Path: "discard",
Format: AuditFormatJSON,
Options: []Option{WithFileMode("0007")},
IsTempDirPath: false,
IsErrorExpected: false,
ExpectedPath: "discard",
ExpectedFormat: AuditFormatJSON,
ExpectedPrefix: "",
ExpectedFileMode: defaultFileMode,
},
"prefix": {
IsTempDirPath: true,
Path: "vault.log",
Format: AuditFormatJSON,
Options: []Option{WithFileMode("0007"), WithPrefix("bleep")},
IsErrorExpected: false,
ExpectedPrefix: "bleep",
ExpectedFormat: AuditFormatJSON,
ExpectedFileMode: 0o007,
},
}
for name, tc := range tests {
name := name
tc := tc
t.Run(name, func(t *testing.T) {
// t.Parallel()
// If we need a real directory as a path we can use a temp dir.
// but we should keep track of it for comparison in the new sink.
var tempDir string
tempPath := tc.Path
if tc.IsTempDirPath {
tempDir = t.TempDir()
tempPath = filepath.Join(tempDir, tempPath)
}
sink, err := NewAuditFileSink(tempPath, tc.Format, tc.Options...)
switch {
case tc.IsErrorExpected:
require.Error(t, err)
require.EqualError(t, err, tc.ExpectedErrorMessage)
require.Nil(t, sink)
default:
require.NoError(t, err)
require.NotNil(t, sink)
// Assert properties are correct.
require.Equal(t, tc.ExpectedPrefix, sink.prefix)
require.Equal(t, tc.ExpectedFormat, sink.format)
require.Equal(t, tc.ExpectedFileMode, sink.fileMode)
switch {
case tc.IsTempDirPath:
require.Equal(t, tempPath, sink.path)
default:
require.Equal(t, tc.ExpectedPath, sink.path)
}
}
})
}
}
// TestAuditFileSink_Reopen tests that the sink reopens files as expected when requested to.
// stdout and discard paths are ignored.
// see: https://developer.hashicorp.com/vault/docs/audit/file#file_path
func TestAuditFileSink_Reopen(t *testing.T) {
tests := map[string]struct {
Path string
IsTempDirPath bool
ShouldCreateFile bool
Options []Option
IsErrorExpected bool
ExpectedErrorMessage string
ExpectedFileMode os.FileMode
}{
// Should be ignored by Reopen
"discard": {
Path: "discard",
},
// Should be ignored by Reopen
"stdout": {
Path: "stdout",
},
"permission-denied": {
Path: "/tmp/vault/test/foo.log",
IsErrorExpected: true,
ExpectedErrorMessage: "event.(AuditFileSink).open: unable to create file \"/tmp/vault/test/foo.log\": mkdir /tmp/vault/test: permission denied",
},
"happy": {
Path: "vault.log",
IsTempDirPath: true,
ExpectedFileMode: os.FileMode(defaultFileMode),
},
"filemode-existing": {
Path: "vault.log",
IsTempDirPath: true,
ShouldCreateFile: true,
Options: []Option{WithFileMode("0000")},
ExpectedFileMode: os.FileMode(defaultFileMode),
},
}
for name, tc := range tests {
name := name
tc := tc
t.Run(name, func(t *testing.T) {
t.Parallel()
// If we need a real directory as a path we can use a temp dir.
// but we should keep track of it for comparison in the new sink.
var tempDir string
tempPath := tc.Path
if tc.IsTempDirPath {
tempDir = t.TempDir()
tempPath = filepath.Join(tempDir, tc.Path)
}
// If the file mode is 0 then we will need a pre-created file to stat.
// Only do this for paths that are not 'special keywords'
if tc.ShouldCreateFile && tc.Path != discard && tc.Path != stdout {
f, err := os.OpenFile(tempPath, os.O_CREATE, defaultFileMode)
require.NoError(t, err)
defer func() {
err = os.Remove(f.Name())
require.NoError(t, err)
}()
}
sink, err := NewAuditFileSink(tempPath, AuditFormatJSON, tc.Options...)
require.NoError(t, err)
require.NotNil(t, sink)
err = sink.Reopen()
switch {
case tc.IsErrorExpected:
require.Error(t, err)
require.EqualError(t, err, tc.ExpectedErrorMessage)
case tempPath == discard:
require.NoError(t, err)
case tempPath == stdout:
require.NoError(t, err)
default:
require.NoError(t, err)
info, err := os.Stat(tempPath)
require.NoError(t, err)
require.NotNil(t, info)
require.Equal(t, tc.ExpectedFileMode, info.Mode())
}
})
}
}
// TestAuditFileSink_Process ensures that Process behaves as expected.
func TestAuditFileSink_Process(t *testing.T) {
tests := map[string]struct {
Path string
Format auditFormat
Data *logical.LogInput
ShouldIgnoreFormat bool
ShouldUseNilEvent bool
IsErrorExpected bool
ExpectedErrorMessage string
}{
"discard": {
Path: discard,
Format: AuditFormatJSON,
Data: &logical.LogInput{Request: &logical.Request{ID: "123"}},
IsErrorExpected: false,
},
"stdout": {
Path: stdout,
Format: AuditFormatJSON,
Data: &logical.LogInput{Request: &logical.Request{ID: "123"}},
IsErrorExpected: false,
},
"no-formatted-data": {
Path: "/dev/null",
Format: AuditFormatJSON,
Data: &logical.LogInput{Request: &logical.Request{ID: "123"}},
ShouldIgnoreFormat: true,
IsErrorExpected: true,
ExpectedErrorMessage: "event.(AuditFileSink).Process: unable to retrieve event formatted as \"json\"",
},
"nil": {
Path: "/dev/null",
Format: AuditFormatJSON,
Data: &logical.LogInput{Request: &logical.Request{ID: "123"}},
ShouldUseNilEvent: true,
IsErrorExpected: true,
ExpectedErrorMessage: "event.(AuditFileSink).Process: event is nil: invalid parameter",
},
}
for name, tc := range tests {
name := name
tc := tc
t.Run(name, func(t *testing.T) {
// Setup a formatter
cfg := vaultaudit.FormatterConfig{}
ss := newStaticSalt(t)
formatter, err := NewAuditFormatterJSON(cfg, ss)
require.NoError(t, err)
require.NotNil(t, formatter)
// Setup a sink
sink, err := NewAuditFileSink(tc.Path, tc.Format)
require.NoError(t, err)
require.NotNil(t, sink)
// Generate a fake event
ctx := namespace.RootContext(nil)
event := fakeJSONAuditEvent(t, AuditRequest, tc.Data)
require.NotNil(t, event)
// Finesse the event into the correct shape.
event, err = formatter.Process(ctx, event)
require.NoError(t, err)
require.NotNil(t, event)
// Some conditional stuff 'per test' to exercise different parts of Process.
if tc.ShouldIgnoreFormat {
delete(event.Formatted, tc.Format.String())
}
if tc.ShouldUseNilEvent {
event = nil
}
// The actual exercising of the sink.
event, err = sink.Process(ctx, event)
switch {
case tc.IsErrorExpected:
require.Error(t, err)
require.EqualError(t, err, tc.ExpectedErrorMessage)
require.Nil(t, event)
default:
require.NoError(t, err)
require.Nil(t, event)
}
})
}
}
// BenchmarkAuditFileSink_Process benchmarks the AuditFormatterJSON and then AuditFileSink calling Process.
// This should replicate the original benchmark testing which used to perform both of these roles together.
func BenchmarkAuditFileSink_Process(b *testing.B) {
// Base input
in := &logical.LogInput{
Auth: &logical.Auth{
ClientToken: "foo",
Accessor: "bar",
EntityID: "foobarentity",
DisplayName: "testtoken",
NoDefaultPolicy: true,
Policies: []string{"root"},
TokenType: logical.TokenTypeService,
},
Request: &logical.Request{
Operation: logical.UpdateOperation,
Path: "/foo",
Connection: &logical.Connection{
RemoteAddr: "127.0.0.1",
},
WrapInfo: &logical.RequestWrapInfo{
TTL: 60 * time.Second,
},
Headers: map[string][]string{
"foo": {"bar"},
},
},
}
ctx := namespace.RootContext(nil)
// Create the formatter node.
cfg := vaultaudit.FormatterConfig{}
ss := newStaticSalt(b)
formatter, err := NewAuditFormatterJSON(cfg, ss)
require.NoError(b, err)
require.NotNil(b, formatter)
// Create the sink node.
sink, err := NewAuditFileSink("/dev/null", AuditFormatJSON)
require.NoError(b, err)
require.NotNil(b, sink)
// Generate the event
event := fakeJSONAuditEvent(b, AuditRequest, in)
require.NotNil(b, event)
b.ResetTimer()
b.RunParallel(func(pb *testing.PB) {
for pb.Next() {
event, err = formatter.Process(ctx, event)
if err != nil {
panic(err)
}
_, err := sink.Process(ctx, event)
if err != nil {
panic(err)
}
}
})
}

135
internal/observability/event/event_type_audit.go
View File

@@ -1,135 +0,0 @@
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: MPL-2.0
package event
import (
"fmt"
"time"
"github.com/hashicorp/vault/sdk/logical"
)
// Audit subtypes.
const (
AuditRequest auditSubtype = "AuditRequest"
AuditResponse auditSubtype = "AuditResponse"
)
// Audit formats.
const (
AuditFormatJSON auditFormat = "json"
AuditFormatJSONx auditFormat = "jsonx"
)
// auditVersion defines the version of audit events.
const auditVersion = "v0.1"
// auditSubtype defines the type of audit event.
type auditSubtype string
// auditFormat defines types of format audit events support.
type auditFormat string
// audit is the audit event.
type audit struct {
ID string `json:"id"`
Version string `json:"version"`
Subtype auditSubtype `json:"subtype"` // the subtype of the audit event.
Timestamp time.Time `json:"timestamp"`
Data *logical.LogInput `json:"data"`
RequiredFormat auditFormat `json:"format"`
}
// newAudit should be used to create an audit event.
// auditSubtype and auditFormat are needed for audit.
// It will use the supplied options, generate an ID if required, and validate the event.
func newAudit(opt ...Option) (*audit, error) {
const op = "event.newAudit"
opts, err := getOpts(opt...)
if err != nil {
return nil, fmt.Errorf("%s: error applying options: %w", op, err)
}
if opts.withID == "" {
var err error
opts.withID, err = NewID(string(AuditType))
if err != nil {
return nil, fmt.Errorf("%s: error creating ID for event: %w", op, err)
}
}
audit := &audit{
ID: opts.withID,
Version: auditVersion,
Subtype: opts.withSubtype,
Timestamp: opts.withNow,
RequiredFormat: opts.withFormat,
}
if err := audit.validate(); err != nil {
return nil, fmt.Errorf("%s: %w", op, err)
}
return audit, nil
}
// validate attempts to ensure the event has the basic requirements of the event type configured.
func (a *audit) validate() error {
const op = "event.(audit).validate"
if a == nil {
return fmt.Errorf("%s: audit is nil: %w", op, ErrInvalidParameter)
}
if a.ID == "" {
return fmt.Errorf("%s: missing ID: %w", op, ErrInvalidParameter)
}
if a.Version != auditVersion {
return fmt.Errorf("%s: audit version unsupported: %w", op, ErrInvalidParameter)
}
if a.Timestamp.IsZero() {
return fmt.Errorf("%s: audit timestamp cannot be the zero time instant: %w", op, ErrInvalidParameter)
}
err := a.Subtype.validate()
if err != nil {
return fmt.Errorf("%s: %w", op, err)
}
err = a.RequiredFormat.validate()
if err != nil {
return fmt.Errorf("%s: %w", op, err)
}
return nil
}
// validate ensures that auditSubtype is one of the set of allowed event subtypes.
func (t auditSubtype) validate() error {
const op = "event.(auditSubtype).validate"
switch t {
case AuditRequest, AuditResponse:
return nil
default:
return fmt.Errorf("%s: '%s' is not a valid event subtype: %w", op, t, ErrInvalidParameter)
}
}
// validate ensures that auditFormat is one of the set of allowed event formats.
func (f auditFormat) validate() error {
const op = "event.(auditFormat).validate"
switch f {
case AuditFormatJSON, AuditFormatJSONx:
return nil
default:
return fmt.Errorf("%s: '%s' is not a valid format: %w", op, f, ErrInvalidParameter)
}
}
// String returns the string version of an auditFormat.
func (f auditFormat) String() string {
return string(f)
}

293
internal/observability/event/event_type_audit_test.go
View File

@@ -1,293 +0,0 @@
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: MPL-2.0
package event
import (
"testing"
"time"
"github.com/stretchr/testify/require"
)
// TestAuditEvent_New exercises the newAudit func to create audit events.
func TestAuditEvent_New(t *testing.T) {
tests := map[string]struct {
Options []Option
IsErrorExpected bool
ExpectedErrorMessage string
ExpectedID string
ExpectedFormat auditFormat
ExpectedSubtype auditSubtype
ExpectedTimestamp time.Time
IsNowExpected bool
}{
"nil": {
Options: nil,
IsErrorExpected: true,
ExpectedErrorMessage: "event.newAudit: event.(audit).validate: event.(auditSubtype).validate: '' is not a valid event subtype: invalid parameter",
},
"empty-option": {
Options: []Option{},
IsErrorExpected: true,
ExpectedErrorMessage: "event.newAudit: event.(audit).validate: event.(auditSubtype).validate: '' is not a valid event subtype: invalid parameter",
},
"bad-id": {
Options: []Option{WithID("")},
IsErrorExpected: true,
ExpectedErrorMessage: "event.newAudit: error applying options: id cannot be empty",
},
"good": {
Options: []Option{
WithID("audit_123"),
WithFormat(string(AuditFormatJSON)),
WithSubtype(string(AuditResponse)),
WithNow(time.Date(2023, time.July, 4, 12, 3, 0, 0, time.Local)),
},
IsErrorExpected: false,
ExpectedID: "audit_123",
ExpectedTimestamp: time.Date(2023, time.July, 4, 12, 3, 0, 0, time.Local),
ExpectedSubtype: AuditResponse,
ExpectedFormat: AuditFormatJSON,
},
"good-no-time": {
Options: []Option{
WithID("audit_123"),
WithFormat(string(AuditFormatJSON)),
WithSubtype(string(AuditResponse)),
},
IsErrorExpected: false,
ExpectedID: "audit_123",
ExpectedSubtype: AuditResponse,
ExpectedFormat: AuditFormatJSON,
IsNowExpected: true,
},
}
for name, tc := range tests {
name := name
tc := tc
t.Run(name, func(t *testing.T) {
t.Parallel()
audit, err := newAudit(tc.Options...)
switch {
case tc.IsErrorExpected:
require.Error(t, err)
require.EqualError(t, err, tc.ExpectedErrorMessage)
require.Nil(t, audit)
default:
require.NoError(t, err)
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))
require.False(t, audit.Timestamp.IsZero())
default:
require.Equal(t, tc.ExpectedTimestamp, audit.Timestamp)
}
}
})
}
}
// TestAuditEvent_Validate exercises the validation for an audit event.
func TestAuditEvent_Validate(t *testing.T) {
tests := map[string]struct {
Value *audit
IsErrorExpected bool
ExpectedErrorMessage string
}{
"nil": {
Value: nil,
IsErrorExpected: true,
ExpectedErrorMessage: "event.(audit).validate: audit is nil: invalid parameter",
},
"default": {
Value: &audit{},
IsErrorExpected: true,
ExpectedErrorMessage: "event.(audit).validate: missing ID: invalid parameter",
},
"id-empty": {
Value: &audit{
ID: "",
Version: auditVersion,
Subtype: AuditRequest,
Timestamp: time.Now(),
Data: nil,
RequiredFormat: AuditFormatJSON,
},
IsErrorExpected: true,
ExpectedErrorMessage: "event.(audit).validate: missing ID: invalid parameter",
},
"version-fiddled": {
Value: &audit{
ID: "audit_123",
Version: "magic-v2",
Subtype: AuditRequest,
Timestamp: time.Now(),
Data: nil,
RequiredFormat: AuditFormatJSON,
},
IsErrorExpected: true,
ExpectedErrorMessage: "event.(audit).validate: audit version unsupported: invalid parameter",
},
"subtype-fiddled": {
Value: &audit{
ID: "audit_123",
Version: auditVersion,
Subtype: auditSubtype("moon"),
Timestamp: time.Now(),
Data: nil,
RequiredFormat: AuditFormatJSON,
},
IsErrorExpected: true,
ExpectedErrorMessage: "event.(audit).validate: event.(auditSubtype).validate: 'moon' is not a valid event subtype: invalid parameter",
},
"format-fiddled": {
Value: &audit{
ID: "audit_123",
Version: auditVersion,
Subtype: AuditResponse,
Timestamp: time.Now(),
Data: nil,
RequiredFormat: auditFormat("blah"),
},
IsErrorExpected: true,
ExpectedErrorMessage: "event.(audit).validate: event.(auditFormat).validate: 'blah' is not a valid format: invalid parameter",
},
"default-time": {
Value: &audit{
ID: "audit_123",
Version: auditVersion,
Subtype: AuditResponse,
Timestamp: time.Time{},
Data: nil,
RequiredFormat: AuditFormatJSON,
},
IsErrorExpected: true,
ExpectedErrorMessage: "event.(audit).validate: audit timestamp cannot be the zero time instant: invalid parameter",
},
"valid": {
Value: &audit{
ID: "audit_123",
Version: auditVersion,
Subtype: AuditResponse,
Timestamp: time.Now(),
Data: nil,
RequiredFormat: AuditFormatJSON,
},
IsErrorExpected: false,
},
}
for name, tc := range tests {
name := name
tc := tc
t.Run(name, func(t *testing.T) {
t.Parallel()
err := tc.Value.validate()
switch {
case tc.IsErrorExpected:
require.Error(t, err)
require.EqualError(t, err, tc.ExpectedErrorMessage)
default:
require.NoError(t, err)
}
})
}
}
// TestAuditEvent_Validate_Subtype exercises the validation for an audit event's subtype.
func TestAuditEvent_Validate_Subtype(t *testing.T) {
tests := map[string]struct {
Value string
IsErrorExpected bool
ExpectedErrorMessage string
}{
"empty": {
Value: "",
IsErrorExpected: true,
ExpectedErrorMessage: "event.(auditSubtype).validate: '' is not a valid event subtype: invalid parameter",
},
"unsupported": {
Value: "foo",
IsErrorExpected: true,
ExpectedErrorMessage: "event.(auditSubtype).validate: 'foo' is not a valid event subtype: invalid parameter",
},
"request": {
Value: "AuditRequest",
IsErrorExpected: false,
},
"response": {
Value: "AuditResponse",
IsErrorExpected: false,
},
}
for name, tc := range tests {
name := name
tc := tc
t.Run(name, func(t *testing.T) {
t.Parallel()
err := auditSubtype(tc.Value).validate()
switch {
case tc.IsErrorExpected:
require.Error(t, err)
require.EqualError(t, err, tc.ExpectedErrorMessage)
default:
require.NoError(t, err)
}
})
}
}
// TestAuditEvent_Validate_Format exercises the validation for an audit event's format.
func TestAuditEvent_Validate_Format(t *testing.T) {
tests := map[string]struct {
Value string
IsErrorExpected bool
ExpectedErrorMessage string
}{
"empty": {
Value: "",
IsErrorExpected: true,
ExpectedErrorMessage: "event.(auditFormat).validate: '' is not a valid format: invalid parameter",
},
"unsupported": {
Value: "foo",
IsErrorExpected: true,
ExpectedErrorMessage: "event.(auditFormat).validate: 'foo' is not a valid format: invalid parameter",
},
"json": {
Value: "json",
IsErrorExpected: false,
},
"jsonx": {
Value: "jsonx",
IsErrorExpected: false,
},
}
for name, tc := range tests {
name := name
tc := tc
t.Run(name, func(t *testing.T) {
t.Parallel()
err := auditFormat(tc.Value).validate()
switch {
case tc.IsErrorExpected:
require.Error(t, err)
require.EqualError(t, err, tc.ExpectedErrorMessage)
default:
require.NoError(t, err)
}
})
}
}

104
internal/observability/event/node_formatter_audit_json.go
View File

@@ -1,104 +0,0 @@
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: MPL-2.0
package event
import (
"context"
"fmt"
vaultaudit "github.com/hashicorp/vault/audit"
"github.com/hashicorp/vault/sdk/helper/jsonutil"
"github.com/hashicorp/eventlogger"
)
var _ eventlogger.Node = (*AuditFormatterJSON)(nil)
// AuditFormatterJSON represents the formatter node which is used to handle
// formatting audit events as JSON.
type AuditFormatterJSON struct {
config vaultaudit.FormatterConfig
format auditFormat
formatter vaultaudit.Formatter
}
// NewAuditFormatterJSON should be used to create an AuditFormatterJSON.
func NewAuditFormatterJSON(config vaultaudit.FormatterConfig, salter vaultaudit.Salter) (*AuditFormatterJSON, error) {
const op = "event.NewAuditFormatterJSON"
f, err := vaultaudit.NewAuditFormatter(salter)
if err != nil {
return nil, fmt.Errorf("%s: unable to create new JSON audit formatter: %w", op, err)
}
jsonFormatter := &AuditFormatterJSON{
format: AuditFormatJSON,
config: config,
formatter: f,
}
return jsonFormatter, nil
}
// Reopen is a no-op for a formatter node.
func (_ *AuditFormatterJSON) Reopen() error {
return nil
}
// Type describes the type of this node (formatter).
func (_ *AuditFormatterJSON) Type() eventlogger.NodeType {
return eventlogger.NodeTypeFormatter
}
// Process will attempt to parse the incoming event data into a corresponding
// audit request/response entry which is serialized to JSON and stored within the event.
func (f *AuditFormatterJSON) Process(ctx context.Context, e *eventlogger.Event) (*eventlogger.Event, error) {
const op = "event.(AuditFormatterJSON).Process"
select {
case <-ctx.Done():
return nil, ctx.Err()
default:
}
if e == nil {
return nil, fmt.Errorf("%s: event is nil: %w", op, ErrInvalidParameter)
}
a, ok := e.Payload.(*audit)
if !ok {
return nil, fmt.Errorf("%s: cannot parse event payload: %w", op, ErrInvalidParameter)
}
var formatted []byte
switch a.Subtype {
case AuditRequest:
entry, err := f.formatter.FormatRequest(ctx, f.config, a.Data)
if err != nil {
return nil, fmt.Errorf("%s: unable to parse request from audit event: %w", op, err)
}
formatted, err = jsonutil.EncodeJSON(entry)
if err != nil {
return nil, fmt.Errorf("%s: unable to format request: %w", op, err)
}
case AuditResponse:
entry, err := f.formatter.FormatResponse(ctx, f.config, a.Data)
if err != nil {
return nil, fmt.Errorf("%s: unable to parse response from audit event: %w", op, err)
}
formatted, err = jsonutil.EncodeJSON(entry)
if err != nil {
return nil, fmt.Errorf("%s: unable to format response: %w", op, err)
}
default:
return nil, fmt.Errorf("%s: unknown audit event subtype: %q", op, a.Subtype)
}
e.FormattedAs(f.format.String(), formatted)
return e, nil
}

242
internal/observability/event/node_formatter_audit_json_test.go
View File

@@ -1,242 +0,0 @@
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: MPL-2.0
package event
import (
"context"
"testing"
"time"
"github.com/hashicorp/vault/helper/namespace"
"github.com/hashicorp/vault/sdk/logical"
vaultaudit "github.com/hashicorp/vault/audit"
"github.com/hashicorp/vault/sdk/helper/salt"
"github.com/hashicorp/eventlogger"
"github.com/stretchr/testify/require"
)
// fakeJSONAuditEvent will return a new fake event containing audit data based
// on the specified auditSubtype and logical.LogInput.
func fakeJSONAuditEvent(tb testing.TB, subtype auditSubtype, input *logical.LogInput) *eventlogger.Event {
tb.Helper()
date := time.Date(2023, time.July, 11, 15, 49, 10, 0o0, time.Local)
auditEvent, err := newAudit(
WithID("123"),
WithSubtype(string(subtype)),
WithFormat(string(AuditFormatJSON)),
WithNow(date),
)
require.NoError(tb, err)
require.NotNil(tb, auditEvent)
require.Equal(tb, "123", auditEvent.ID)
require.Equal(tb, "v0.1", auditEvent.Version)
require.Equal(tb, AuditFormatJSON, auditEvent.RequiredFormat)
require.Equal(tb, subtype, auditEvent.Subtype)
require.Equal(tb, date, auditEvent.Timestamp)
auditEvent.Data = input
e := &eventlogger.Event{
Type: eventlogger.EventType(AuditType),
CreatedAt: auditEvent.Timestamp,
Formatted: make(map[string][]byte),
Payload: auditEvent,
}
return e
}
// newStaticSalt returns a new staticSalt for use in testing.
func newStaticSalt(tb testing.TB) *staticSalt {
s, err := salt.NewSalt(context.Background(), nil, nil)
require.NoError(tb, err)
return &staticSalt{salt: s}
}
// staticSalt is a struct which can be used to obtain a static salt.
// a salt must be assigned when the struct is initialized.
type staticSalt struct {
salt *salt.Salt
}
// Salt returns the static salt and no error.
func (s *staticSalt) Salt(_ context.Context) (*salt.Salt, error) {
return s.salt, nil
}
// TestNewAuditFormatterJSON ensures we can create new AuditFormatterJSONX structs.
func TestNewAuditFormatterJSON(t *testing.T) {
tests := map[string]struct {
UseStaticSalt bool
IsErrorExpected bool
ExpectedErrorMessage string
}{
"nil-salter": {
UseStaticSalt: false,
IsErrorExpected: true,
ExpectedErrorMessage: "event.NewAuditFormatterJSON: unable to create new JSON audit formatter: cannot create a new audit formatter with nil salter",
},
"static-salter": {
UseStaticSalt: true,
IsErrorExpected: false,
},
}
for name, tc := range tests {
name := name
tc := tc
t.Run(name, func(t *testing.T) {
t.Parallel()
cfg := vaultaudit.FormatterConfig{}
var ss vaultaudit.Salter
if tc.UseStaticSalt {
ss = newStaticSalt(t)
}
f, err := NewAuditFormatterJSON(cfg, ss)
switch {
case tc.IsErrorExpected:
require.Error(t, err)
require.EqualError(t, err, tc.ExpectedErrorMessage)
require.Nil(t, f)
default:
require.NoError(t, err)
require.NotNil(t, f)
}
})
}
}
// TestAuditFormatterJSONX_Reopen ensures that we do no get an error when calling Reopen.
func TestAuditFormatterJSON_Reopen(t *testing.T) {
ss := newStaticSalt(t)
cfg := vaultaudit.FormatterConfig{}
f, err := NewAuditFormatterJSON(cfg, ss)
require.NoError(t, err)
require.NotNil(t, f)
require.NoError(t, f.Reopen())
}
// TestAuditFormatterJSONX_Type ensures that the node is a 'formatter' type.
func TestAuditFormatterJSON_Type(t *testing.T) {
ss := newStaticSalt(t)
cfg := vaultaudit.FormatterConfig{}
f, err := NewAuditFormatterJSON(cfg, ss)
require.NoError(t, err)
require.NotNil(t, f)
require.Equal(t, eventlogger.NodeTypeFormatter, f.Type())
}
// TestAuditFormatterJSON_Process attempts to run the Process method to convert
// the logical.LogInput within an audit event to JSON (AuditRequestEntry or AuditResponseEntry).
func TestAuditFormatterJSON_Process(t *testing.T) {
tests := map[string]struct {
IsErrorExpected bool
ExpectedErrorMessage string
Subtype auditSubtype
Data *logical.LogInput
RootNamespace bool
}{
"request-no-data": {
IsErrorExpected: true,
ExpectedErrorMessage: "event.(AuditFormatterJSON).Process: unable to parse request from audit event: request to request-audit a nil request",
Subtype: AuditRequest,
Data: nil,
},
"response-no-data": {
IsErrorExpected: true,
ExpectedErrorMessage: "event.(AuditFormatterJSON).Process: unable to parse response from audit event: request to response-audit a nil request",
Subtype: AuditResponse,
Data: nil,
},
"request-basic-input": {
IsErrorExpected: true,
ExpectedErrorMessage: "event.(AuditFormatterJSON).Process: unable to parse request from audit event: request to request-audit a nil request",
Subtype: AuditRequest,
Data: &logical.LogInput{Type: "magic"},
},
"response-basic-input": {
IsErrorExpected: true,
ExpectedErrorMessage: "event.(AuditFormatterJSON).Process: unable to parse response from audit event: request to response-audit a nil request",
Subtype: AuditResponse,
Data: &logical.LogInput{Type: "magic"},
},
"request-basic-input-and-request-no-ns": {
IsErrorExpected: true,
ExpectedErrorMessage: "event.(AuditFormatterJSON).Process: unable to parse request from audit event: no namespace",
Subtype: AuditRequest,
Data: &logical.LogInput{Request: &logical.Request{ID: "123"}},
},
"response-basic-input-and-request-no-ns": {
IsErrorExpected: true,
ExpectedErrorMessage: "event.(AuditFormatterJSON).Process: unable to parse response from audit event: no namespace",
Subtype: AuditResponse,
Data: &logical.LogInput{Request: &logical.Request{ID: "123"}},
},
"request-basic-input-and-request-with-ns": {
IsErrorExpected: false,
Subtype: AuditRequest,
Data: &logical.LogInput{Request: &logical.Request{ID: "123"}},
RootNamespace: true,
},
"response-basic-input-and-request-with-ns": {
IsErrorExpected: false,
Subtype: AuditResponse,
Data: &logical.LogInput{Request: &logical.Request{ID: "123"}},
RootNamespace: true,
},
}
for name, tc := range tests {
name := name
tc := tc
t.Run(name, func(t *testing.T) {
t.Parallel()
e := fakeJSONAuditEvent(t, tc.Subtype, tc.Data)
require.NotNil(t, e)
ss := newStaticSalt(t)
cfg := vaultaudit.FormatterConfig{}
f, err := NewAuditFormatterJSON(cfg, ss)
require.NoError(t, err)
require.NotNil(t, f)
var ctx context.Context
switch {
case tc.RootNamespace:
ctx = namespace.RootContext(context.Background())
default:
ctx = context.Background()
}
processed, err := f.Process(ctx, e)
b, found := e.Format(string(AuditFormatJSON))
switch {
case tc.IsErrorExpected:
require.Error(t, err)
require.EqualError(t, err, tc.ExpectedErrorMessage)
require.Nil(t, processed)
require.False(t, found)
require.Nil(t, b)
default:
require.NoError(t, err)
require.NotNil(t, processed)
require.True(t, found)
require.NotNil(t, b)
}
})
}
}

75
internal/observability/event/node_formatter_audit_jsonx.go
View File

@@ -1,75 +0,0 @@
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: MPL-2.0
package event
import (
"context"
"fmt"
"github.com/hashicorp/eventlogger"
"github.com/jefferai/jsonx"
)
var _ eventlogger.Node = (*AuditFormatterJSONx)(nil)
// AuditFormatterJSONx represents a formatter node which will Process JSON to JSONx format.
type AuditFormatterJSONx struct {
format auditFormat
}
// NewAuditFormatterJSONx creates a formatter node which can be used to format
// incoming events to JSONx.
// This formatter node requires that a AuditFormatterJSON node exists earlier
// in the pipeline and will attempt to access the JSON encoded data stored by that
// formatter node.
func NewAuditFormatterJSONx() *AuditFormatterJSONx {
return &AuditFormatterJSONx{format: AuditFormatJSONx}
}
// Reopen is a no-op for this formatter node.
func (_ *AuditFormatterJSONx) Reopen() error {
return nil
}
// Type describes the type of this node.
func (_ *AuditFormatterJSONx) Type() eventlogger.NodeType {
return eventlogger.NodeTypeFormatter
}
// Process will attempt to retrieve pre-formatted JSON stored within the event
// and re-encode the data to JSONx.
func (f *AuditFormatterJSONx) Process(ctx context.Context, e *eventlogger.Event) (*eventlogger.Event, error) {
const op = "event.(AuditFormatterJSONx).Process"
select {
case <-ctx.Done():
return nil, ctx.Err()
default:
}
if e == nil {
return nil, fmt.Errorf("%s: event is nil: %w", op, ErrInvalidParameter)
}
// We expect that JSON has already been parsed for this event.
jsonBytes, ok := e.Format(AuditFormatJSON.String())
if !ok {
return nil, fmt.Errorf("%s: pre-formatted JSON required but not found: %w", op, ErrInvalidParameter)
}
if jsonBytes == nil {
return nil, fmt.Errorf("%s: pre-formatted JSON required but was nil: %w", op, ErrInvalidParameter)
}
xmlBytes, err := jsonx.EncodeJSONBytes(jsonBytes)
if err != nil {
return nil, fmt.Errorf("%s: unable to encode JSONx using JSON data: %w", op, err)
}
if xmlBytes == nil {
return nil, fmt.Errorf("%s: encoded JSONx was nil: %w", op, err)
}
e.FormattedAs(f.format.String(), xmlBytes)
return e, nil
}

138
internal/observability/event/node_formatter_audit_jsonx_test.go
View File

@@ -1,138 +0,0 @@
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: MPL-2.0
package event
import (
"context"
"testing"
"time"
"github.com/hashicorp/vault/sdk/logical"
"github.com/hashicorp/vault/sdk/helper/jsonutil"
"github.com/hashicorp/eventlogger"
"github.com/stretchr/testify/require"
)
// fakeJSONxAuditEvent will return a new fake event containing audit data based
// on the specified auditSubtype and logical.LogInput.
func fakeJSONxAuditEvent(t *testing.T, subtype auditSubtype, input *logical.LogInput) *eventlogger.Event {
t.Helper()
date := time.Date(2023, time.July, 11, 15, 49, 10, 0, time.Local)
auditEvent, err := newAudit(
WithID("123"),
WithSubtype(string(subtype)),
WithFormat(string(AuditFormatJSONx)),
WithNow(date),
)
require.NoError(t, err)
require.NotNil(t, auditEvent)
require.Equal(t, "123", auditEvent.ID)
require.Equal(t, "v0.1", auditEvent.Version)
require.Equal(t, AuditFormatJSONx, auditEvent.RequiredFormat)
require.Equal(t, subtype, auditEvent.Subtype)
require.Equal(t, date, auditEvent.Timestamp)
auditEvent.Data = input
e := &eventlogger.Event{
Type: eventlogger.EventType(AuditType),
CreatedAt: auditEvent.Timestamp,
Formatted: make(map[string][]byte),
Payload: auditEvent,
}
return e
}
// TestNewAuditFormatterJSONx ensures we can create new AuditFormatterJSONx structs.
func TestNewAuditFormatterJSONx(t *testing.T) {
f := NewAuditFormatterJSONx()
require.NotNil(t, f)
}
// TestAuditFormatterJSONx_Reopen ensures that we do no get an error when calling Reopen.
func TestAuditFormatterJSONx_Reopen(t *testing.T) {
require.NoError(t, NewAuditFormatterJSONx().Reopen())
}
// TestAuditFormatterJSONx_Type ensures that the node is a 'formatter' type.
func TestAuditFormatterJSONx_Type(t *testing.T) {
require.Equal(t, eventlogger.NodeTypeFormatter, NewAuditFormatterJSONx().Type())
}
// TestAuditFormatterJSONx_Process attempts to run the Process method to convert
// pre-formatted JSON to XML (JSONx).
func TestAuditFormatterJSONx_Process(t *testing.T) {
tests := map[string]struct {
IsErrorExpected bool
ExpectedErrorMessage string
Subtype auditSubtype
Data *logical.LogInput
}{
"request-no-formatted-json": {
IsErrorExpected: true,
ExpectedErrorMessage: "event.(AuditFormatterJSONx).Process: pre-formatted JSON required but not found: invalid parameter",
Subtype: AuditRequest,
Data: nil,
},
"response-no-formatted-json": {
IsErrorExpected: true,
ExpectedErrorMessage: "event.(AuditFormatterJSONx).Process: pre-formatted JSON required but not found: invalid parameter",
Subtype: AuditResponse,
Data: nil,
},
"request-basic-json": {
IsErrorExpected: false,
Subtype: AuditRequest,
Data: &logical.LogInput{Type: "magic"},
},
"response-basic-json": {
IsErrorExpected: false,
Subtype: AuditResponse,
Data: &logical.LogInput{Type: "magic"},
},
}
for name, tc := range tests {
name := name
tc := tc
t.Run(name, func(t *testing.T) {
t.Parallel()
e := fakeJSONxAuditEvent(t, tc.Subtype, tc.Data)
require.NotNil(t, e)
// If we have data specified, then encode it and store as a format.
// This is faking the behavior of the JSON formatter node which is a
// pre-req for JSONx formatter node.
if tc.Data != nil {
jsonBytes, err := jsonutil.EncodeJSON(tc.Data)
require.NoError(t, err)
require.NotNil(t, jsonBytes)
e.FormattedAs(string(AuditFormatJSON), jsonBytes)
}
processed, err := NewAuditFormatterJSONx().Process(context.Background(), e)
b, found := e.Format(string(AuditFormatJSONx))
switch {
case tc.IsErrorExpected:
require.Error(t, err)
require.EqualError(t, err, tc.ExpectedErrorMessage)
require.Nil(t, processed)
require.False(t, found)
require.Nil(t, b)
default:
require.NoError(t, err)
require.NotNil(t, processed)
require.True(t, found)
require.NotNil(t, b)
}
})
}
}

153
internal/observability/event/options.go
View File

@@ -19,21 +19,18 @@ import (
// Option is how Options are passed as arguments.
type Option func(*options) error
// options are used to represent configuration for an Event.
// Options are used to represent configuration for an Event.
type options struct {
withID string
withNow time.Time
withSubtype auditSubtype
withFormat auditFormat
withFileMode *os.FileMode
withPrefix string
withFacility string
withTag string
withSocketType string
withMaxDuration time.Duration
withFileMode *os.FileMode
}
// getDefaultOptions returns options with their default values.
// getDefaultOptions returns Options with their default values.
func getDefaultOptions() options {
return options{
withNow: time.Now(),
@@ -44,7 +41,7 @@ func getDefaultOptions() options {
}
}
// getOpts applies all the supplied Option and returns configured options.
// getOpts applies all the supplied Option and returns configured Options.
// Each Option is applied in the order it appears in the argument list, so it is
// possible to supply the same Option numerous times and the 'last write wins'.
func getOpts(opt ...Option) (options, error) {
@@ -110,51 +107,72 @@ func WithNow(now time.Time) Option {
}
}
// WithSubtype provides an option to represent the subtype.
func WithSubtype(subtype string) Option {
// WithFacility provides an Option to represent a 'facility' for a syslog sink.
func WithFacility(facility string) Option {
return func(o *options) error {
s := strings.TrimSpace(subtype)
if s == "" {
return errors.New("subtype cannot be empty")
facility = strings.TrimSpace(facility)
if facility != "" {
o.withFacility = facility
}
parsed := auditSubtype(s)
err := parsed.validate()
if err != nil {
return err
}
o.withSubtype = parsed
return nil
}
}
// WithFormat provides an option to represent event format.
func WithFormat(format string) Option {
// WithTag provides an Option to represent a 'tag' for a syslog sink.
func WithTag(tag string) Option {
return func(o *options) error {
f := strings.TrimSpace(format)
if f == "" {
return errors.New("format cannot be empty")
tag = strings.TrimSpace(tag)
if tag != "" {
o.withTag = tag
}
parsed := auditFormat(f)
err := parsed.validate()
if err != nil {
return err
}
o.withFormat = parsed
return nil
}
}
// WithFileMode provides an option to represent a file mode for a file sink.
// Supplying an empty string or whitespace will prevent this option from being
// WithSocketType provides an Option to represent the socket type for a socket sink.
func WithSocketType(socketType string) Option {
return func(o *options) error {
socketType = strings.TrimSpace(socketType)
if socketType != "" {
o.withSocketType = socketType
}
return nil
}
}
// WithMaxDuration provides an Option to represent the max duration for writing to a socket.
func WithMaxDuration(duration string) Option {
return func(o *options) error {
duration = strings.TrimSpace(duration)
if duration == "" {
return nil
}
parsed, err := parseutil.ParseDurationSecond(duration)
if err != nil {
return err
}
o.withMaxDuration = parsed
return nil
}
}
// WithFileMode provides an Option to represent a file mode for a file sink.
// Supplying an empty string or whitespace will prevent this Option from being
// applied, but it will not return an error in those circumstances.
func WithFileMode(mode string) Option {
return func(o *options) error {
// If supplied file mode is empty, just return early without setting anything.
// We can assume that this option was called by something that didn't
// We can assume that this Option was called by something that didn't
// parse the incoming value, perhaps from a config map etc.
mode = strings.TrimSpace(mode)
if mode == "" {
@@ -176,70 +194,3 @@ func WithFileMode(mode string) Option {
return nil
}
}
// WithPrefix provides an option to represent a prefix for a file sink.
func WithPrefix(prefix string) Option {
return func(o *options) error {
o.withPrefix = prefix
return nil
}
}
// WithFacility provides an option to represent a 'facility' for a syslog sink.
func WithFacility(facility string) Option {
return func(o *options) error {
facility = strings.TrimSpace(facility)
if facility != "" {
o.withFacility = facility
}
return nil
}
}
// WithTag provides an option to represent a 'tag' for a syslog sink.
func WithTag(tag string) Option {
return func(o *options) error {
tag = strings.TrimSpace(tag)
if tag != "" {
o.withTag = tag
}
return nil
}
}
// WithSocketType provides an option to represent the socket type for a socket sink.
func WithSocketType(socketType string) Option {
return func(o *options) error {
socketType = strings.TrimSpace(socketType)
if socketType != "" {
o.withSocketType = socketType
}
return nil
}
}
// WithMaxDuration provides an option to represent the max duration for writing to a socket sink.
func WithMaxDuration(duration string) Option {
return func(o *options) error {
duration = strings.TrimSpace(duration)
if duration == "" {
return nil
}
parsed, err := parseutil.ParseDurationSecond(duration)
if err != nil {
return err
}
o.withMaxDuration = parsed
return nil
}
}

336
internal/observability/event/options_test.go
View File

@@ -11,106 +11,6 @@ import (
"github.com/stretchr/testify/require"
)
// TestOptions_WithFormat exercises WithFormat option to ensure it performs as expected.
func TestOptions_WithFormat(t *testing.T) {
tests := map[string]struct {
Value string
IsErrorExpected bool
ExpectedErrorMessage string
ExpectedValue auditFormat
}{
"empty": {
Value: "",
IsErrorExpected: true,
ExpectedErrorMessage: "format cannot be empty",
},
"whitespace": {
Value: " ",
IsErrorExpected: true,
ExpectedErrorMessage: "format cannot be empty",
},
"invalid-test": {
Value: "test",
IsErrorExpected: true,
ExpectedErrorMessage: "event.(auditFormat).validate: 'test' is not a valid format: invalid parameter",
},
"valid-json": {
Value: "json",
IsErrorExpected: false,
ExpectedValue: AuditFormatJSON,
},
"valid-jsonx": {
Value: "jsonx",
IsErrorExpected: false,
ExpectedValue: AuditFormatJSONx,
},
}
for name, tc := range tests {
name := name
tc := tc
t.Run(name, func(t *testing.T) {
t.Parallel()
options := &options{}
applyOption := WithFormat(tc.Value)
err := applyOption(options)
switch {
case tc.IsErrorExpected:
require.Error(t, err)
require.EqualError(t, err, tc.ExpectedErrorMessage)
default:
require.NoError(t, err)
require.Equal(t, tc.ExpectedValue, options.withFormat)
}
})
}
}
// TestOptions_WithSubtype exercises WithSubtype option to ensure it performs as expected.
func TestOptions_WithSubtype(t *testing.T) {
tests := map[string]struct {
Value string
IsErrorExpected bool
ExpectedErrorMessage string
ExpectedValue auditSubtype
}{
"empty": {
Value: "",
IsErrorExpected: true,
ExpectedErrorMessage: "subtype cannot be empty",
},
"whitespace": {
Value: " ",
IsErrorExpected: true,
ExpectedErrorMessage: "subtype cannot be empty",
},
"valid": {
Value: "AuditResponse",
IsErrorExpected: false,
ExpectedValue: AuditResponse,
},
}
for name, tc := range tests {
name := name
tc := tc
t.Run(name, func(t *testing.T) {
t.Parallel()
options := &options{}
applyOption := WithSubtype(tc.Value)
err := applyOption(options)
switch {
case tc.IsErrorExpected:
require.Error(t, err)
require.EqualError(t, err, tc.ExpectedErrorMessage)
default:
require.NoError(t, err)
require.Equal(t, tc.ExpectedValue, options.withSubtype)
}
})
}
}
// TestOptions_WithNow exercises WithNow option to ensure it performs as expected.
func TestOptions_WithNow(t *testing.T) {
tests := map[string]struct {
@@ -137,16 +37,16 @@ func TestOptions_WithNow(t *testing.T) {
t.Run(name, func(t *testing.T) {
t.Parallel()
options := &options{}
opts := &options{}
applyOption := WithNow(tc.Value)
err := applyOption(options)
err := applyOption(opts)
switch {
case tc.IsErrorExpected:
require.Error(t, err)
require.EqualError(t, err, tc.ExpectedErrorMessage)
default:
require.NoError(t, err)
require.Equal(t, tc.ExpectedValue, options.withNow)
require.Equal(t, tc.ExpectedValue, opts.withNow)
}
})
}
@@ -197,7 +97,103 @@ func TestOptions_WithID(t *testing.T) {
}
}
// TestOptions_WithFacility exercises WithFacility option to ensure it performs as expected.
// TestOptions_Default exercises getDefaultOptions to assert the default values.
func TestOptions_Default(t *testing.T) {
opts := getDefaultOptions()
require.NotNil(t, opts)
require.True(t, time.Now().After(opts.withNow))
require.False(t, opts.withNow.IsZero())
require.Equal(t, "AUTH", opts.withFacility)
require.Equal(t, "vault", opts.withTag)
require.Equal(t, 2*time.Second, opts.withMaxDuration)
}
// TestOptions_Opts exercises getOpts with various Option values.
func TestOptions_Opts(t *testing.T) {
tests := map[string]struct {
opts []Option
IsErrorExpected bool
ExpectedErrorMessage string
ExpectedID string
IsNowExpected bool
ExpectedNow time.Time
}{
"nil-options": {
opts: nil,
IsErrorExpected: false,
IsNowExpected: true,
},
"empty-options": {
opts: []Option{},
IsErrorExpected: false,
IsNowExpected: true,
},
"with-multiple-valid-id": {
opts: []Option{
WithID("qwerty"),
WithID("juan"),
},
IsErrorExpected: false,
ExpectedID: "juan",
IsNowExpected: true,
},
"with-multiple-valid-now": {
opts: []Option{
WithNow(time.Date(2023, time.July, 4, 12, 3, 0, 0, time.Local)),
WithNow(time.Date(2023, time.July, 4, 13, 3, 0, 0, time.Local)),
},
IsErrorExpected: false,
ExpectedNow: time.Date(2023, time.July, 4, 13, 3, 0, 0, time.Local),
IsNowExpected: false,
},
"with-multiple-valid-then-invalid-now": {
opts: []Option{
WithNow(time.Date(2023, time.July, 4, 12, 3, 0, 0, time.Local)),
WithNow(time.Time{}),
},
IsErrorExpected: true,
ExpectedErrorMessage: "cannot specify 'now' to be the zero time instant",
},
"with-multiple-valid-options": {
opts: []Option{
WithID("qwerty"),
WithNow(time.Date(2023, time.July, 4, 12, 3, 0, 0, time.Local)),
},
IsErrorExpected: false,
ExpectedID: "qwerty",
ExpectedNow: time.Date(2023, time.July, 4, 12, 3, 0, 0, time.Local),
},
}
for name, tc := range tests {
name := name
tc := tc
t.Run(name, func(t *testing.T) {
t.Parallel()
opts, err := getOpts(tc.opts...)
switch {
case tc.IsErrorExpected:
require.Error(t, err)
require.EqualError(t, err, tc.ExpectedErrorMessage)
default:
require.NotNil(t, opts)
require.NoError(t, err)
require.Equal(t, tc.ExpectedID, opts.withID)
switch {
case tc.IsNowExpected:
require.True(t, time.Now().After(opts.withNow))
require.False(t, opts.withNow.IsZero())
default:
require.Equal(t, tc.ExpectedNow, opts.withNow)
}
}
})
}
}
// TestOptions_WithFacility exercises WithFacility Option to ensure it performs as expected.
func TestOptions_WithFacility(t *testing.T) {
tests := map[string]struct {
Value string
@@ -235,7 +231,7 @@ func TestOptions_WithFacility(t *testing.T) {
}
}
// TestOptions_WithTag exercises WithTag option to ensure it performs as expected.
// TestOptions_WithTag exercises WithTag Option to ensure it performs as expected.
func TestOptions_WithTag(t *testing.T) {
tests := map[string]struct {
Value string
@@ -273,7 +269,7 @@ func TestOptions_WithTag(t *testing.T) {
}
}
// TestOptions_WithSocketType exercises WithSocketType option to ensure it performs as expected.
// TestOptions_WithSocketType exercises WithSocketType Option to ensure it performs as expected.
func TestOptions_WithSocketType(t *testing.T) {
tests := map[string]struct {
Value string
@@ -311,7 +307,7 @@ func TestOptions_WithSocketType(t *testing.T) {
}
}
// TestOptions_WithMaxDuration exercises WithMaxDuration option to ensure it performs as expected.
// TestOptions_WithMaxDuration exercises WithMaxDuration Option to ensure it performs as expected.
func TestOptions_WithMaxDuration(t *testing.T) {
tests := map[string]struct {
Value string
@@ -365,7 +361,7 @@ func TestOptions_WithMaxDuration(t *testing.T) {
}
}
// TestOptions_WithFileMode exercises WithFileMode option to ensure it performs as expected.
// TestOptions_WithFileMode exercises WithFileMode Option to ensure it performs as expected.
func TestOptions_WithFileMode(t *testing.T) {
tests := map[string]struct {
Value string
@@ -417,7 +413,7 @@ func TestOptions_WithFileMode(t *testing.T) {
require.NoError(t, err)
switch {
case tc.IsNilExpected:
// Optional option 'not supplied' (i.e. was whitespace/empty string)
// Optional Option 'not supplied' (i.e. was whitespace/empty string)
require.Nil(t, options.withFileMode)
default:
// Dereference the pointer, so we can examine the file mode.
@@ -427,125 +423,3 @@ func TestOptions_WithFileMode(t *testing.T) {
})
}
}
// TestOptions_Default exercises getDefaultOptions to assert the default values.
func TestOptions_Default(t *testing.T) {
opts := getDefaultOptions()
require.NotNil(t, opts)
require.True(t, time.Now().After(opts.withNow))
require.False(t, opts.withNow.IsZero())
require.Equal(t, "AUTH", opts.withFacility)
require.Equal(t, "vault", opts.withTag)
require.Equal(t, 2*time.Second, opts.withMaxDuration)
}
// TestOptions_Opts exercises getOpts with various Option values.
func TestOptions_Opts(t *testing.T) {
tests := map[string]struct {
opts []Option
IsErrorExpected bool
ExpectedErrorMessage string
ExpectedID string
ExpectedSubtype auditSubtype
ExpectedFormat auditFormat
IsNowExpected bool
ExpectedNow time.Time
}{
"nil-options": {
opts: nil,
IsErrorExpected: false,
IsNowExpected: true,
},
"empty-options": {
opts: []Option{},
IsErrorExpected: false,
IsNowExpected: true,
},
"with-multiple-valid-id": {
opts: []Option{
WithID("qwerty"),
WithID("juan"),
},
IsErrorExpected: false,
ExpectedID: "juan",
IsNowExpected: true,
},
"with-multiple-valid-subtype": {
opts: []Option{
WithSubtype("AuditRequest"),
WithSubtype("AuditResponse"),
},
IsErrorExpected: false,
ExpectedSubtype: AuditResponse,
IsNowExpected: true,
},
"with-multiple-valid-format": {
opts: []Option{
WithFormat("json"),
WithFormat("jsonx"),
},
IsErrorExpected: false,
ExpectedFormat: AuditFormatJSONx,
IsNowExpected: true,
},
"with-multiple-valid-now": {
opts: []Option{
WithNow(time.Date(2023, time.July, 4, 12, 3, 0, 0, time.Local)),
WithNow(time.Date(2023, time.July, 4, 13, 3, 0, 0, time.Local)),
},
IsErrorExpected: false,
ExpectedNow: time.Date(2023, time.July, 4, 13, 3, 0, 0, time.Local),
IsNowExpected: false,
},
"with-multiple-valid-then-invalid-now": {
opts: []Option{
WithNow(time.Date(2023, time.July, 4, 12, 3, 0, 0, time.Local)),
WithNow(time.Time{}),
},
IsErrorExpected: true,
ExpectedErrorMessage: "cannot specify 'now' to be the zero time instant",
},
"with-multiple-valid-options": {
opts: []Option{
WithID("qwerty"),
WithSubtype("AuditRequest"),
WithFormat("json"),
WithNow(time.Date(2023, time.July, 4, 12, 3, 0, 0, time.Local)),
},
IsErrorExpected: false,
ExpectedID: "qwerty",
ExpectedSubtype: AuditRequest,
ExpectedFormat: AuditFormatJSON,
ExpectedNow: time.Date(2023, time.July, 4, 12, 3, 0, 0, time.Local),
},
}
for name, tc := range tests {
name := name
tc := tc
t.Run(name, func(t *testing.T) {
t.Parallel()
opts, err := getOpts(tc.opts...)
switch {
case tc.IsErrorExpected:
require.Error(t, err)
require.EqualError(t, err, tc.ExpectedErrorMessage)
default:
require.NotNil(t, opts)
require.NoError(t, err)
require.Equal(t, tc.ExpectedID, opts.withID)
require.Equal(t, tc.ExpectedSubtype, opts.withSubtype)
require.Equal(t, tc.ExpectedFormat, opts.withFormat)
switch {
case tc.IsNowExpected:
require.True(t, time.Now().After(opts.withNow))
require.False(t, opts.withNow.IsZero())
default:
require.Equal(t, tc.ExpectedNow, opts.withNow)
}
}
})
}
}

226
internal/observability/event/sink_file.go Normal file
View File

@@ -0,0 +1,226 @@
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: MPL-2.0
package event
import (
"bytes"
"context"
"fmt"
"io"
"os"
"path/filepath"
"strings"
"sync"
"github.com/hashicorp/eventlogger"
)
// defaultFileMode is the default file permissions (read/write for everyone).
const (
defaultFileMode = 0o600
devnull = "/dev/null"
)
// FileSink is a sink node which handles writing events to file.
type FileSink struct {
file *os.File
fileLock sync.RWMutex
fileMode os.FileMode
path string
requiredFormat string
}
// NewFileSink should be used to create a new FileSink.
// Accepted options: WithFileMode.
func NewFileSink(path string, format string, opt ...Option) (*FileSink, error) {
const op = "event.NewFileSink"
// Parse and check path
p := strings.TrimSpace(path)
if p == "" {
return nil, fmt.Errorf("%s: path is required", op)
}
opts, err := getOpts(opt...)
if err != nil {
return nil, fmt.Errorf("%s: error applying options: %w", op, err)
}
mode := os.FileMode(defaultFileMode)
// If we got an optional file mode supplied and our path isn't a special keyword
// then we should use the supplied file mode, or maintain the existing file mode.
switch {
case path == devnull:
case opts.withFileMode == nil:
case *opts.withFileMode == 0: // Maintain the existing file's mode when set to "0000".
fileInfo, err := os.Stat(path)
if err != nil {
return nil, fmt.Errorf("%s: unable to determine existing file mode: %w", op, err)
}
mode = fileInfo.Mode()
default:
mode = *opts.withFileMode
}
sink := &FileSink{
file: nil,
fileLock: sync.RWMutex{},
fileMode: mode,
requiredFormat: format,
path: p,
}
// Ensure that the file can be successfully opened for writing;
// otherwise it will be too late to catch later without problems
// (ref: https://github.com/hashicorp/vault/issues/550)
if err := sink.open(); err != nil {
return nil, fmt.Errorf("%s: sanity check failed; unable to open %q for writing: %w", op, path, err)
}
return sink, nil
}
// Process handles writing the event to the file sink.
func (f *FileSink) Process(ctx context.Context, e *eventlogger.Event) (*eventlogger.Event, error) {
const op = "event.(FileSink).Process"
select {
case <-ctx.Done():
return nil, ctx.Err()
default:
}
if e == nil {
return nil, fmt.Errorf("%s: event is nil: %w", op, ErrInvalidParameter)
}
// '/dev/null' path means we just do nothing and pretend we're done.
if f.path == devnull {
return nil, nil
}
formatted, found := e.Format(f.requiredFormat)
if !found {
return nil, fmt.Errorf("%s: unable to retrieve event formatted as %q", op, f.requiredFormat)
}
err := f.log(formatted)
if err != nil {
return nil, fmt.Errorf("%s: error writing file for sink: %w", op, err)
}
// return nil for the event to indicate the pipeline is complete.
return nil, nil
}
// Reopen handles closing and reopening the file.
func (f *FileSink) Reopen() error {
const op = "event.(FileSink).Reopen"
// '/dev/null' path means we just do nothing and pretend we're done.
if f.path == devnull {
return nil
}
f.fileLock.Lock()
defer f.fileLock.Unlock()
if f.file == nil {
return f.open()
}
err := f.file.Close()
// Set to nil here so that even if we error out, on the next access open() will be tried.
f.file = nil
if err != nil {
return fmt.Errorf("%s: unable to close file for re-opening on sink: %w", op, err)
}
return f.open()
}
// Type describes the type of this node (sink).
func (_ *FileSink) Type() eventlogger.NodeType {
return eventlogger.NodeTypeSink
}
// open attempts to open a file at the sink's path, with the sink's fileMode permissions
// if one is not already open.
// It doesn't have any locking and relies on calling functions of FileSink to
// handle this (e.g. log and Reopen methods).
func (f *FileSink) open() error {
const op = "event.(FileSink).open"
if f.file != nil {
return nil
}
if err := os.MkdirAll(filepath.Dir(f.path), f.fileMode); err != nil {
return fmt.Errorf("%s: unable to create file %q: %w", op, f.path, err)
}
var err error
f.file, err = os.OpenFile(f.path, os.O_APPEND|os.O_WRONLY|os.O_CREATE, f.fileMode)
if err != nil {
return fmt.Errorf("%s: unable to open file for sink: %w", op, err)
}
// Change the file mode in case the log file already existed.
// We special case '/dev/null' since we can't chmod it, and bypass if the mode is zero.
switch f.path {
case devnull:
default:
if f.fileMode != 0 {
err = os.Chmod(f.path, f.fileMode)
if err != nil {
return fmt.Errorf("%s: unable to change file %q permissions '%v' for sink: %w", op, f.path, f.fileMode, err)
}
}
}
return nil
}
// log writes the buffer to the file.
// It acquires a lock on the file to do this.
func (f *FileSink) log(data []byte) error {
const op = "event.(FileSink).log"
f.fileLock.Lock()
defer f.fileLock.Unlock()
reader := bytes.NewReader(data)
if err := f.open(); err != nil {
return fmt.Errorf("%s: unable to open file for sink: %w", op, err)
}
if _, err := reader.WriteTo(f.file); err == nil {
return nil
}
// Otherwise, opportunistically try to re-open the FD, once per call (1 retry attempt).
err := f.file.Close()
if err != nil {
return fmt.Errorf("%s: unable to close file for sink: %w", op, err)
}
f.file = nil
if err := f.open(); err != nil {
return fmt.Errorf("%s: unable to re-open file for sink: %w", op, err)
}
_, err = reader.Seek(0, io.SeekStart)
if err != nil {
return fmt.Errorf("%s: unable to seek to start of file for sink: %w", op, err)
}
_, err = reader.WriteTo(f.file)
if err != nil {
return fmt.Errorf("%s: unable to re-write to file for sink: %w", op, err)
}
return nil
}

301
internal/observability/event/sink_file_test.go Normal file
View File

@@ -0,0 +1,301 @@
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: MPL-2.0
package event
import (
"os"
"path/filepath"
"testing"
"time"
"github.com/hashicorp/vault/helper/namespace"
"github.com/hashicorp/eventlogger"
"github.com/stretchr/testify/require"
)
// TestFileSink_Type ensures that the node is a 'sink' type.
func TestFileSink_Type(t *testing.T) {
f, err := NewFileSink(filepath.Join(t.TempDir(), "vault.log"), "json")
require.NoError(t, err)
require.NotNil(t, f)
require.Equal(t, eventlogger.NodeTypeSink, f.Type())
}
// TestNewFileSink tests creation of an AuditFileSink.
func TestNewFileSink(t *testing.T) {
tests := map[string]struct {
ShouldUseAbsolutePath bool // Path should contain the filename if temp dir is true
Path string
Format string
Options []Option
IsErrorExpected bool
ExpectedErrorMessage string
// Expected values of AuditFileSink
ExpectedFileMode os.FileMode
ExpectedFormat string
ExpectedPath string
ExpectedPrefix string
}{
"default-values": {
ShouldUseAbsolutePath: true,
IsErrorExpected: true,
ExpectedErrorMessage: "event.NewFileSink: path is required",
},
"spacey-path": {
ShouldUseAbsolutePath: true,
Path: " ",
Format: "json",
IsErrorExpected: true,
ExpectedErrorMessage: "event.NewFileSink: path is required",
},
"valid-path-and-format": {
Path: "vault.log",
Format: "json",
IsErrorExpected: false,
ExpectedFileMode: defaultFileMode,
ExpectedFormat: "json",
ExpectedPrefix: "",
},
"file-mode-not-default-or-zero": {
Path: "vault.log",
Format: "json",
Options: []Option{WithFileMode("0007")},
IsErrorExpected: false,
ExpectedFormat: "json",
ExpectedPrefix: "",
ExpectedFileMode: 0o007,
},
"prefix": {
Path: "vault.log",
Format: "json",
Options: []Option{WithFileMode("0007")},
IsErrorExpected: false,
ExpectedPrefix: "bleep",
ExpectedFormat: "json",
ExpectedFileMode: 0o007,
},
}
for name, tc := range tests {
name := name
tc := tc
t.Run(name, func(t *testing.T) {
// t.Parallel()
// If we need a real directory as a path we can use a temp dir.
// but we should keep track of it for comparison in the new sink.
var tempDir string
tempPath := tc.Path
if !tc.ShouldUseAbsolutePath {
tempDir = t.TempDir()
tempPath = filepath.Join(tempDir, tempPath)
}
sink, err := NewFileSink(tempPath, tc.Format, tc.Options...)
switch {
case tc.IsErrorExpected:
require.Error(t, err)
require.EqualError(t, err, tc.ExpectedErrorMessage)
require.Nil(t, sink)
default:
require.NoError(t, err)
require.NotNil(t, sink)
// Assert properties are correct.
require.Equal(t, tc.ExpectedFormat, sink.requiredFormat)
require.Equal(t, tc.ExpectedFileMode, sink.fileMode)
switch {
case tc.ShouldUseAbsolutePath:
require.Equal(t, tc.ExpectedPath, sink.path)
default:
require.Equal(t, tempPath, sink.path)
}
}
})
}
}
// TestFileSink_Reopen tests that the sink reopens files as expected when requested to.
// stdout and discard paths are ignored.
// see: https://developer.hashicorp.com/vault/docs/audit/file#file_path
func TestFileSink_Reopen(t *testing.T) {
tests := map[string]struct {
Path string
ShouldUseAbsolutePath bool
ShouldCreateFile bool
ShouldIgnoreFileMode bool
Options []Option
IsErrorExpected bool
ExpectedErrorMessage string
ExpectedFileMode os.FileMode
}{
// Should be ignored by Reopen
"devnull": {
Path: "/dev/null",
ShouldUseAbsolutePath: true,
ShouldIgnoreFileMode: true,
},
"happy": {
Path: "vault.log",
ExpectedFileMode: os.FileMode(defaultFileMode),
},
"filemode-existing": {
Path: "vault.log",
ShouldCreateFile: true,
Options: []Option{WithFileMode("0000")},
ExpectedFileMode: os.FileMode(defaultFileMode),
},
}
for name, tc := range tests {
name := name
tc := tc
t.Run(name, func(t *testing.T) {
t.Parallel()
// If we need a real directory as a path we can use a temp dir.
// but we should keep track of it for comparison in the new sink.
var tempDir string
tempPath := tc.Path
if !tc.ShouldUseAbsolutePath {
tempDir = t.TempDir()
tempPath = filepath.Join(tempDir, tc.Path)
}
// If the file mode is 0 then we will need a pre-created file to stat.
// Only do this for paths that are not 'special keywords'
if tc.ShouldCreateFile && tc.Path != devnull {
f, err := os.OpenFile(tempPath, os.O_CREATE, defaultFileMode)
require.NoError(t, err)
defer func() {
err = os.Remove(f.Name())
require.NoError(t, err)
}()
}
sink, err := NewFileSink(tempPath, "json", tc.Options...)
require.NoError(t, err)
require.NotNil(t, sink)
err = sink.Reopen()
switch {
case tc.IsErrorExpected:
require.Error(t, err)
require.EqualError(t, err, tc.ExpectedErrorMessage)
default:
require.NoError(t, err)
info, err := os.Stat(tempPath)
require.NoError(t, err)
require.NotNil(t, info)
if !tc.ShouldIgnoreFileMode {
require.Equal(t, tc.ExpectedFileMode, info.Mode())
}
}
})
}
}
// TestFileSink_Process ensures that Process behaves as expected.
func TestFileSink_Process(t *testing.T) {
tests := map[string]struct {
ShouldUseAbsolutePath bool
Path string
ShouldCreateFile bool
Format string
ShouldIgnoreFormat bool
Data string
ShouldUseNilEvent bool
IsErrorExpected bool
ExpectedErrorMessage string
}{
"devnull": {
ShouldUseAbsolutePath: true,
Path: devnull,
Format: "json",
Data: "foo",
IsErrorExpected: false,
},
"no-formatted-data": {
ShouldCreateFile: true,
Path: "juan.log",
Format: "json",
Data: "foo",
ShouldIgnoreFormat: true,
IsErrorExpected: true,
ExpectedErrorMessage: "event.(FileSink).Process: unable to retrieve event formatted as \"json\"",
},
"nil": {
Path: "foo.log",
Format: "json",
Data: "foo",
ShouldUseNilEvent: true,
IsErrorExpected: true,
ExpectedErrorMessage: "event.(FileSink).Process: event is nil: invalid parameter",
},
}
for name, tc := range tests {
name := name
tc := tc
t.Run(name, func(t *testing.T) {
// Temp dir for most testing unless we're trying to test an error
var tempDir string
tempPath := tc.Path
if !tc.ShouldUseAbsolutePath {
tempDir = t.TempDir()
tempPath = filepath.Join(tempDir, tc.Path)
}
// Create a file if we will need it there before Process kicks off.
if tc.ShouldCreateFile && tc.Path != devnull {
f, err := os.OpenFile(tempPath, os.O_CREATE, defaultFileMode)
require.NoError(t, err)
defer func() {
err = os.Remove(f.Name())
require.NoError(t, err)
}()
}
// Set up a sink
sink, err := NewFileSink(tempPath, tc.Format)
require.NoError(t, err)
require.NotNil(t, sink)
// Generate a fake event
ctx := namespace.RootContext(nil)
event := &eventlogger.Event{
Type: "audit",
CreatedAt: time.Now(),
Formatted: make(map[string][]byte),
Payload: struct{ ID string }{ID: "123"},
}
if !tc.ShouldIgnoreFormat {
event.FormattedAs(tc.Format, []byte(tc.Data))
}
if tc.ShouldUseNilEvent {
event = nil
}
// The actual exercising of the sink.
event, err = sink.Process(ctx, event)
switch {
case tc.IsErrorExpected:
require.Error(t, err)
require.EqualError(t, err, tc.ExpectedErrorMessage)
require.Nil(t, event)
default:
require.NoError(t, err)
require.Nil(t, event)
}
})
}
}

34
internal/observability/event/sink_noop.go Normal file
View File

@@ -0,0 +1,34 @@
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: MPL-2.0
package event
import (
"context"
"github.com/hashicorp/eventlogger"
)
// NoopSink is a sink node which handles ignores everything.
type NoopSink struct{}
// NewNoopSink should be used to create a new NoopSink.
func NewNoopSink() *NoopSink {
return &NoopSink{}
}
// Process is a no-op and always returns nil event and nil error.
func (_ *NoopSink) Process(ctx context.Context, _ *eventlogger.Event) (*eventlogger.Event, error) {
// return nil for the event to indicate the pipeline is complete.
return nil, nil
}
// Reopen is a no-op and always returns nil.
func (_ *NoopSink) Reopen() error {
return nil
}
// Type describes the type of this node (sink).
func (_ *NoopSink) Type() eventlogger.NodeType {
return eventlogger.NodeTypeSink
}

66
internal/observability/event/audit_sink_socket.go → internal/observability/event/sink_socket.go
View File

@@ -15,41 +15,41 @@ import (
"github.com/hashicorp/eventlogger"
)
// AuditSocketSink is a sink node which handles writing audit events to socket.
type AuditSocketSink struct {
format auditFormat
address string
socketType string
maxDuration time.Duration
socketLock sync.RWMutex
connection net.Conn
// SocketSink is a sink node which handles writing events to socket.
type SocketSink struct {
requiredFormat string
address string
socketType string
maxDuration time.Duration
socketLock sync.RWMutex
connection net.Conn
}
// NewAuditSocketSink should be used to create a new AuditSocketSink.
// NewSocketSink should be used to create a new SocketSink.
// Accepted options: WithMaxDuration and WithSocketType.
func NewAuditSocketSink(format auditFormat, address string, opt ...Option) (*AuditSocketSink, error) {
const op = "event.NewAuditSocketSink"
func NewSocketSink(format string, address string, opt ...Option) (*SocketSink, error) {
const op = "event.NewSocketSink"
opts, err := getOpts(opt...)
if err != nil {
return nil, fmt.Errorf("%s: error applying options: %w", op, err)
}
sink := &AuditSocketSink{
format: format,
address: address,
socketType: opts.withSocketType,
maxDuration: opts.withMaxDuration,
socketLock: sync.RWMutex{},
connection: nil,
sink := &SocketSink{
requiredFormat: format,
address: address,
socketType: opts.withSocketType,
maxDuration: opts.withMaxDuration,
socketLock: sync.RWMutex{},
connection: nil,
}
return sink, nil
}
// Process handles writing the event to the socket.
func (s *AuditSocketSink) Process(ctx context.Context, e *eventlogger.Event) (*eventlogger.Event, error) {
const op = "event.(AuditSocketSink).Process"
func (s *SocketSink) Process(ctx context.Context, e *eventlogger.Event) (*eventlogger.Event, error) {
const op = "event.(SocketSink).Process"
select {
case <-ctx.Done():
@@ -64,9 +64,9 @@ func (s *AuditSocketSink) Process(ctx context.Context, e *eventlogger.Event) (*e
return nil, fmt.Errorf("%s: event is nil: %w", op, ErrInvalidParameter)
}
formatted, found := e.Format(s.format.String())
formatted, found := e.Format(s.requiredFormat)
if !found {
return nil, fmt.Errorf("%s: unable to retrieve event formatted as %q", op, s.format)
return nil, fmt.Errorf("%s: unable to retrieve event formatted as %q", op, s.requiredFormat)
}
// Try writing and return early if successful.
@@ -95,8 +95,8 @@ func (s *AuditSocketSink) Process(ctx context.Context, e *eventlogger.Event) (*e
}
// Reopen handles reopening the connection for the socket sink.
func (s *AuditSocketSink) Reopen() error {
const op = "event.(AuditSocketSink).Reopen"
func (s *SocketSink) Reopen() error {
const op = "event.(SocketSink).Reopen"
s.socketLock.Lock()
defer s.socketLock.Unlock()
@@ -110,13 +110,13 @@ func (s *AuditSocketSink) Reopen() error {
}
// Type describes the type of this node (sink).
func (s *AuditSocketSink) Type() eventlogger.NodeType {
func (_ *SocketSink) Type() eventlogger.NodeType {
return eventlogger.NodeTypeSink
}
// connect attempts to establish a connection using the socketType and address.
func (s *AuditSocketSink) connect(ctx context.Context) error {
const op = "event.(AuditSocketSink).connect"
func (s *SocketSink) connect(ctx context.Context) error {
const op = "event.(SocketSink).connect"
// If we're already connected, we should have disconnected first.
if s.connection != nil {
@@ -138,8 +138,8 @@ func (s *AuditSocketSink) connect(ctx context.Context) error {
}
// disconnect attempts to close and clear an existing connection.
func (s *AuditSocketSink) disconnect() error {
const op = "event.(AuditSocketSink).disconnect"
func (s *SocketSink) disconnect() error {
const op = "event.(SocketSink).disconnect"
// If we're already disconnected, we can return early.
if s.connection == nil {
@@ -156,8 +156,8 @@ func (s *AuditSocketSink) disconnect() error {
}
// reconnect attempts to disconnect and then connect to the configured socketType and address.
func (s *AuditSocketSink) reconnect(ctx context.Context) error {
const op = "event.(AuditSocketSink).reconnect"
func (s *SocketSink) reconnect(ctx context.Context) error {
const op = "event.(SocketSink).reconnect"
err := s.disconnect()
if err != nil {
@@ -173,8 +173,8 @@ func (s *AuditSocketSink) reconnect(ctx context.Context) error {
}
// write attempts to write the specified data using the established connection.
func (s *AuditSocketSink) write(ctx context.Context, data []byte) error {
const op = "event.(AuditSocketSink).write"
func (s *SocketSink) write(ctx context.Context, data []byte) error {
const op = "event.(SocketSink).write"
// Ensure we're connected.
err := s.connect(ctx)

66
internal/observability/event/sink_stdout.go Normal file
View File

@@ -0,0 +1,66 @@
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: MPL-2.0
package event
import (
"context"
"fmt"
"os"
"github.com/hashicorp/eventlogger"
)
var _ eventlogger.Node = (*StdoutSink)(nil)
// StdoutSink is structure that implements the eventlogger.Node interface
// as a Sink node that writes the events to the standard output stream.
type StdoutSink struct {
requiredFormat string
}
// NewStdoutSinkNode creates a new StdoutSink that will persist the events
// it processes using the specified expected format.
func NewStdoutSinkNode(format string) *StdoutSink {
return &StdoutSink{
requiredFormat: format,
}
}
// Process persists the provided eventlogger.Event to the standard output stream.
func (n *StdoutSink) Process(ctx context.Context, event *eventlogger.Event) (*eventlogger.Event, error) {
const op = "event.(StdoutSink).Process"
select {
case <-ctx.Done():
return nil, ctx.Err()
default:
}
if event == nil {
return nil, fmt.Errorf("%s: event is nil: %w", op, ErrInvalidParameter)
}
formattedBytes, found := event.Format(n.requiredFormat)
if !found {
return nil, fmt.Errorf("%s: unable to retrieve event formatted as %q", op, n.requiredFormat)
}
_, err := os.Stdout.Write(formattedBytes)
if err != nil {
return nil, fmt.Errorf("%s: error writing to stdout: %w", op, err)
}
// Return nil, nil to indicate the pipeline is complete.
return nil, nil
}
// Reopen is a no-op for the StdoutSink type.
func (n *StdoutSink) Reopen() error {
return nil
}
// Type returns the eventlogger.NodeTypeSink constant.
func (n *StdoutSink) Type() eventlogger.NodeType {
return eventlogger.NodeTypeSink
}

28
internal/observability/event/audit_sink_syslog.go → internal/observability/event/sink_syslog.go
View File

@@ -12,16 +12,16 @@ import (
"github.com/hashicorp/eventlogger"
)
// AuditSyslogSink is a sink node which handles writing audit events to syslog.
type AuditSyslogSink struct {
format auditFormat
logger gsyslog.Syslogger
// SyslogSink is a sink node which handles writing events to syslog.
type SyslogSink struct {
requiredFormat string
logger gsyslog.Syslogger
}
// NewAuditSyslogSink should be used to create a new AuditSyslogSink.
// NewSyslogSink should be used to create a new SyslogSink.
// Accepted options: WithFacility and WithTag.
func NewAuditSyslogSink(format auditFormat, opt ...Option) (*AuditSyslogSink, error) {
const op = "event.NewAuditSyslogSink"
func NewSyslogSink(format string, opt ...Option) (*SyslogSink, error) {
const op = "event.NewSyslogSink"
opts, err := getOpts(opt...)
if err != nil {
@@ -33,12 +33,12 @@ func NewAuditSyslogSink(format auditFormat, opt ...Option) (*AuditSyslogSink, er
return nil, fmt.Errorf("%s: error creating syslogger: %w", op, err)
}
return &AuditSyslogSink{format: format, logger: logger}, nil
return &SyslogSink{requiredFormat: format, logger: logger}, nil
}
// Process handles writing the event to the syslog.
func (s *AuditSyslogSink) Process(ctx context.Context, e *eventlogger.Event) (*eventlogger.Event, error) {
const op = "event.(AuditSyslogSink).Process"
func (s *SyslogSink) Process(ctx context.Context, e *eventlogger.Event) (*eventlogger.Event, error) {
const op = "event.(SyslogSink).Process"
select {
case <-ctx.Done():
@@ -50,9 +50,9 @@ func (s *AuditSyslogSink) Process(ctx context.Context, e *eventlogger.Event) (*e
return nil, fmt.Errorf("%s: event is nil: %w", op, ErrInvalidParameter)
}
formatted, found := e.Format(s.format.String())
formatted, found := e.Format(s.requiredFormat)
if !found {
return nil, fmt.Errorf("%s: unable to retrieve event formatted as %q", op, s.format)
return nil, fmt.Errorf("%s: unable to retrieve event formatted as %q", op, s.requiredFormat)
}
_, err := s.logger.Write(formatted)
@@ -65,11 +65,11 @@ func (s *AuditSyslogSink) Process(ctx context.Context, e *eventlogger.Event) (*e
}
// Reopen is a no-op for a syslog sink.
func (s *AuditSyslogSink) Reopen() error {
func (_ *SyslogSink) Reopen() error {
return nil
}
// Type describes the type of this node (sink).
func (s *AuditSyslogSink) Type() eventlogger.NodeType {
func (_ *SyslogSink) Type() eventlogger.NodeType {
return eventlogger.NodeTypeSink
}

28
internal/observability/event/sink_type.go
View File

@@ -1,28 +0,0 @@
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: MPL-2.0
package event
import (
"fmt"
)
const (
FileSink SinkType = "file"
SocketSink SinkType = "socket"
SyslogSink SinkType = "syslog"
)
// SinkType defines the type of sink
type SinkType string
// Validate ensures that SinkType is one of the set of allowed sink types.
func (t SinkType) Validate() error {
const op = "event.(SinkType).Validate"
switch t {
case FileSink, SocketSink, SyslogSink:
return nil
default:
return fmt.Errorf("%s: '%s' is not a valid sink type: %w", op, t, ErrInvalidParameter)
}
}

59
internal/observability/event/sink_type_test.go
View File

@@ -1,59 +0,0 @@
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: MPL-2.0
package event
import (
"testing"
"github.com/stretchr/testify/require"
)
// TestSinkType_Validate exercises the validation for a sink type.
func TestSinkType_Validate(t *testing.T) {
tests := map[string]struct {
Value string
IsValid bool
ExpectedError string
}{
"file": {
Value: "file",
IsValid: true,
},
"syslog": {
Value: "syslog",
IsValid: true,
},
"socket": {
Value: "socket",
IsValid: true,
},
"empty": {
Value: "",
IsValid: false,
ExpectedError: "event.(SinkType).Validate: '' is not a valid sink type: invalid parameter",
},
"random": {
Value: "random",
IsValid: false,
ExpectedError: "event.(SinkType).Validate: 'random' is not a valid sink type: invalid parameter",
},
}
for name, tc := range tests {
name := name
tc := tc
t.Run(name, func(t *testing.T) {
t.Parallel()
sinkType := SinkType(tc.Value)
err := sinkType.Validate()
switch {
case tc.IsValid:
require.NoError(t, err)
case !tc.IsValid:
require.Error(t, err)
require.EqualError(t, err, tc.ExpectedError)
}
})
}
}