diff --git a/.golangci.yml b/.golangci.yml new file mode 100644 index 0000000..5bf14b7 --- /dev/null +++ b/.golangci.yml @@ -0,0 +1,57 @@ +run: + timeout: 5m + go: "1.26" + +linters: + enable: + - errcheck + - gosimple + - govet + - ineffassign + - staticcheck + - unused + - gofmt + - gosec + - revive + - misspell + +linters-settings: + gosec: + excludes: + - G104 # Audit errors not checked (errcheck covers this) + revive: + rules: + - name: blank-imports + disabled: false + - name: context-as-argument + disabled: false + - name: dot-imports + disabled: false + - name: error-naming + disabled: false + - name: error-strings + disabled: false + - name: errorf + disabled: false + - name: increment-decrement + disabled: false + - name: indent-error-flow + disabled: false + - name: range + disabled: false + - name: receiver-naming + disabled: false + - name: time-naming + disabled: false + - name: unexported-return + disabled: false + - name: var-declaration + disabled: false + +issues: + exclude-use-default: true + exclude-rules: + - path: _test\.go + linters: + - gosec + - errcheck diff --git a/Makefile b/Makefile index 74ad5fa..2380ccf 100644 --- a/Makefile +++ b/Makefile @@ -10,7 +10,14 @@ fmt: @go fmt ./... lint: - @golangci-lint run + @if command -v golangci-lint >/dev/null 2>&1; then \ + golangci-lint run; \ + elif [ -x "$$(go env GOPATH)/bin/golangci-lint" ]; then \ + "$$(go env GOPATH)/bin/golangci-lint" run; \ + else \ + echo "golangci-lint not found (run 'make dev-setup')"; \ + exit 1; \ + fi cfn-lint: ## Lint CloudFormation templates (requires cfn-lint) @echo "==> Linting CloudFormation templates" diff --git a/convert_test.go b/convert_test.go new file mode 100644 index 0000000..b776d08 --- /dev/null +++ b/convert_test.go @@ -0,0 +1,371 @@ +package main + +import ( + "bytes" + "compress/zlib" + "context" + "encoding/json" + "os" + "testing" + + mpl "github.com/aws/aws-cryptographic-material-providers-library/releases/go/mpl/awscryptographymaterialproviderssmithygenerated" + mpltypes "github.com/aws/aws-cryptographic-material-providers-library/releases/go/mpl/awscryptographymaterialproviderssmithygeneratedtypes" + client "github.com/aws/aws-encryption-sdk/releases/go/encryption-sdk/awscryptographyencryptionsdksmithygenerated" + esdktypes "github.com/aws/aws-encryption-sdk/releases/go/encryption-sdk/awscryptographyencryptionsdksmithygeneratedtypes" + "github.com/parquet-go/parquet-go" +) + +// Scenario 1: Heartbeat filtering — heartbeat events should be skipped +func TestConvertJSONToParquet_HeartbeatFiltering(t *testing.T) { + rawJSON := []byte(`{ + "databaseActivityEventList": [ + {"type": "heartbeat"}, + {"type": "heartbeat"}, + {"type": "activity", "command": "SELECT", "dbUserName": "alice"} + ] + }`) + + filterConfig := &FilterConfig{Drop: []string{}, Query: map[string]interface{}{}} + parquetBytes, err := convertJSONToParquet(rawJSON, filterConfig) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + reader := parquet.NewReader(bytes.NewReader(parquetBytes)) + var rows []DatabaseActivityEvent + for { + var row DatabaseActivityEvent + err := reader.Read(&row) + if err != nil { + break + } + rows = append(rows, row) + } + + if len(rows) != 1 { + t.Fatalf("expected 1 row (heartbeats excluded), got %d", len(rows)) + } + if rows[0].DbUserName != "alice" { + t.Errorf("expected alice, got %s", rows[0].DbUserName) + } +} + +// Scenario 2: Drop fields — configured drop fields should not appear in output +func TestConvertJSONToParquet_DropFields(t *testing.T) { + rawJSON := []byte(`{ + "databaseActivityEventList": [ + {"type": "activity", "command": "SELECT", "dbUserName": "alice", "errorMessage": "sensitive-err"} + ] + }`) + + filterConfig := &FilterConfig{ + Drop: []string{"errorMessage", "command"}, + Query: map[string]interface{}{}, + } + parquetBytes, err := convertJSONToParquet(rawJSON, filterConfig) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + reader := parquet.NewReader(bytes.NewReader(parquetBytes)) + var row DatabaseActivityEvent + err = reader.Read(&row) + if err != nil { + t.Fatalf("failed to read row: %v", err) + } + + if row.ErrorMessage != "" { + t.Errorf("expected errorMessage to be dropped, got %q", row.ErrorMessage) + } + if row.Command != "" { + t.Errorf("expected command to be dropped, got %q", row.Command) + } + if row.DbUserName != "alice" { + t.Errorf("expected dbUserName to be alice, got %q", row.DbUserName) + } +} + +// Scenario 3: Filter query matching — only events matching query are preserved +func TestConvertJSONToParquet_FilterQueryMatching(t *testing.T) { + rawJSON := []byte(`{ + "databaseActivityEventList": [ + {"type": "activity", "command": "SELECT", "dbUserName": "alice"}, + {"type": "activity", "command": "DROP", "dbUserName": "malory"}, + {"type": "activity", "command": "SELECT", "dbUserName": "bob"} + ] + }`) + + filterConfig := &FilterConfig{ + Drop: []string{}, + Query: map[string]interface{}{ + "type": "selector", + "dimension": "command", + "value": "DROP", + }, + } + parquetBytes, err := convertJSONToParquet(rawJSON, filterConfig) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + reader := parquet.NewReader(bytes.NewReader(parquetBytes)) + var rows []DatabaseActivityEvent + for { + var row DatabaseActivityEvent + err := reader.Read(&row) + if err != nil { + break + } + rows = append(rows, row) + } + + if len(rows) != 1 { + t.Fatalf("expected 1 row matching DROP, got %d", len(rows)) + } + if rows[0].DbUserName != "malory" { + t.Errorf("expected malory, got %s", rows[0].DbUserName) + } +} + +// Scenario 4: Malformed JSON — returns error +func TestConvertJSONToParquet_MalformedJSON(t *testing.T) { + filterConfig := &FilterConfig{Drop: []string{}, Query: map[string]interface{}{}} + _, err := convertJSONToParquet([]byte("{not-valid-json"), filterConfig) + if err == nil { + t.Fatal("expected error on malformed JSON, got nil") + } +} + +// Scenario 5: Empty event list — returns valid parquet with 0 rows +func TestConvertJSONToParquet_EmptyEventList(t *testing.T) { + rawJSON := []byte(`{"databaseActivityEventList": []}`) + filterConfig := &FilterConfig{Drop: []string{}, Query: map[string]interface{}{}} + + parquetBytes, err := convertJSONToParquet(rawJSON, filterConfig) + if err != nil { + t.Fatalf("unexpected error on empty event list: %v", err) + } + + reader := parquet.NewReader(bytes.NewReader(parquetBytes)) + var rows []DatabaseActivityEvent + for { + var row DatabaseActivityEvent + if err := reader.Read(&row); err != nil { + break + } + rows = append(rows, row) + } + + if len(rows) != 0 { + t.Fatalf("expected 0 rows, got %d", len(rows)) + } +} + +// Scenario 6: ExitCode formatting — numeric exit codes are converted to string +func TestConvertJSONToParquet_ExitCodeConversion(t *testing.T) { + rawJSON := []byte(`{ + "databaseActivityEventList": [ + {"type": "activity", "exitCode": 0, "dbUserName": "alice"}, + {"type": "activity", "exitCode": "SUCCESS", "dbUserName": "bob"} + ] + }`) + + filterConfig := &FilterConfig{Drop: []string{}, Query: map[string]interface{}{}} + parquetBytes, err := convertJSONToParquet(rawJSON, filterConfig) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + reader := parquet.NewReader(bytes.NewReader(parquetBytes)) + var rows []DatabaseActivityEvent + for { + var row DatabaseActivityEvent + if err := reader.Read(&row); err != nil { + break + } + rows = append(rows, row) + } + + if len(rows) != 2 { + t.Fatalf("expected 2 rows, got %d", len(rows)) + } + if rows[0].ExitCode != "0" { + t.Errorf("expected '0' exit code, got %q", rows[0].ExitCode) + } + if rows[1].ExitCode != "SUCCESS" { + t.Errorf("expected 'SUCCESS' exit code, got %q", rows[1].ExitCode) + } +} + +// TestDecompressZlib_Valid tests round-trip zlib compression and decompression. +func TestDecompressZlib_Valid(t *testing.T) { + originalData := []byte("The quick brown fox jumps over the lazy dog Database Activity Stream") + + var buf bytes.Buffer + writer := zlib.NewWriter(&buf) + _, err := writer.Write(originalData) + if err != nil { + t.Fatalf("failed to write zlib: %v", err) + } + _ = writer.Close() + + decompressed, err := decompressZlib(buf.Bytes()) + if err != nil { + t.Fatalf("failed to decompress: %v", err) + } + + if string(decompressed) != string(originalData) { + t.Errorf("expected %q, got %q", string(originalData), string(decompressed)) + } +} + +// TestDecompressZlib_InvalidInput tests that corrupted data returns error. +func TestDecompressZlib_InvalidInput(t *testing.T) { + corrupted := []byte("not-a-valid-zlib-stream") + _, err := decompressZlib(corrupted) + if err == nil { + t.Fatal("expected error on corrupted zlib stream, got nil") + } +} + +// TestDecompressZlib_EmptyInput tests that empty input returns error. +func TestDecompressZlib_EmptyInput(t *testing.T) { + _, err := decompressZlib([]byte{}) + if err == nil { + t.Fatal("expected error on empty zlib stream, got nil") + } +} + +// TestDecryptAWSEncryptionSDKPayload tests the ESDK decryption path with valid and invalid inputs. +func TestDecryptAWSEncryptionSDKPayload(t *testing.T) { + ctx := context.Background() + rawKey := []byte("01234567890123456789012345678901") // 32-byte key + plaintext := []byte("Hello, AWS Encryption SDK DAS Payload!") + + // 1. Encrypt payload + matProv, err := mpl.NewClient(mpltypes.MaterialProvidersConfig{}) + if err != nil { + t.Fatalf("failed to create mpl client: %v", err) + } + + aesKeyring, err := matProv.CreateRawAesKeyring(ctx, mpltypes.CreateRawAesKeyringInput{ + KeyName: "DataKey", + KeyNamespace: "RawMasterKeyProvider", + WrappingKey: rawKey, + WrappingAlg: mpltypes.AesWrappingAlgAlgAes256GcmIv12Tag16, + }) + if err != nil { + t.Fatalf("failed to create keyring: %v", err) + } + + esdkClient, err := client.NewClient(esdktypes.AwsEncryptionSdkConfig{}) + if err != nil { + t.Fatalf("failed to create esdk client: %v", err) + } + + encOutput, err := esdkClient.Encrypt(ctx, esdktypes.EncryptInput{ + Plaintext: plaintext, + Keyring: aesKeyring, + }) + if err != nil { + t.Fatalf("failed to encrypt: %v", err) + } + + // 2. Decrypt with correct key + decrypted, err := decryptAWSEncryptionSDKPayload(ctx, encOutput.Ciphertext, rawKey) + if err != nil { + t.Fatalf("failed to decrypt valid ciphertext: %v", err) + } + if string(decrypted) != string(plaintext) { + t.Errorf("expected %q, got %q", string(plaintext), string(decrypted)) + } + + // 3. Decrypt with wrong key fails + wrongKey := []byte("11111111111111111111111111111111") + _, err = decryptAWSEncryptionSDKPayload(ctx, encOutput.Ciphertext, wrongKey) + if err == nil { + t.Fatal("expected error decrypting with wrong key, got nil") + } + + // 4. Decrypt malformed ciphertext fails + _, err = decryptAWSEncryptionSDKPayload(ctx, []byte("invalid-ciphertext"), rawKey) + if err == nil { + t.Fatal("expected error decrypting corrupted ciphertext, got nil") + } +} + +// TestFixtures_Unmarshal verifies that all fixtures in testdata/ parse properly. +func TestFixtures_Unmarshal(t *testing.T) { + // S3 notification fixture + s3Data, err := os.ReadFile("testdata/s3_event_notification.json") + if err != nil { + t.Fatalf("failed to read s3 fixture: %v", err) + } + var s3Notif S3EventNotification + if err := json.Unmarshal(s3Data, &s3Notif); err != nil { + t.Fatalf("failed to unmarshal s3 fixture: %v", err) + } + if len(s3Notif.Records) != 1 || s3Notif.Records[0].S3.Bucket.Name != "aws-das-source-bucket" { + t.Errorf("unexpected s3 fixture content: %+v", s3Notif) + } + + // SNS fixture + snsData, err := os.ReadFile("testdata/sns_s3_event.json") + if err != nil { + t.Fatalf("failed to read sns fixture: %v", err) + } + var snsMsg struct { + Type string `json:"Type"` + Message string `json:"Message"` + } + if err := json.Unmarshal(snsData, &snsMsg); err != nil { + t.Fatalf("failed to unmarshal sns fixture: %v", err) + } + var innerS3 S3EventNotification + if err := json.Unmarshal([]byte(snsMsg.Message), &innerS3); err != nil { + t.Fatalf("failed to unmarshal inner S3 notification from SNS: %v", err) + } + if len(innerS3.Records) != 1 { + t.Errorf("expected 1 record in inner S3 notification") + } + + // DAS events fixture + dasData, err := os.ReadFile("testdata/das_events.json") + if err != nil { + t.Fatalf("failed to read das_events fixture: %v", err) + } + filterCfg := &FilterConfig{Drop: []string{}, Query: map[string]interface{}{}} + parquetBytes, err := convertJSONToParquet(dasData, filterCfg) + if err != nil { + t.Fatalf("failed to convert das_events fixture to parquet: %v", err) + } + if len(parquetBytes) == 0 { + t.Errorf("expected non-empty parquet output for das_events fixture") + } + + // Heartbeat-only fixture + hbData, err := os.ReadFile("testdata/das_heartbeat_only.json") + if err != nil { + t.Fatalf("failed to read heartbeat fixture: %v", err) + } + hbBytes, err := convertJSONToParquet(hbData, filterCfg) + if err != nil { + t.Fatalf("failed to convert heartbeat fixture: %v", err) + } + if len(hbBytes) == 0 { + t.Errorf("expected non-empty parquet schema header for heartbeat fixture") + } + + // Empty fixture + emptyData, err := os.ReadFile("testdata/das_empty.json") + if err != nil { + t.Fatalf("failed to read empty fixture: %v", err) + } + emptyBytes, err := convertJSONToParquet(emptyData, filterCfg) + if err != nil { + t.Fatalf("failed to convert empty fixture: %v", err) + } + if len(emptyBytes) == 0 { + t.Errorf("expected non-empty parquet schema header for empty fixture") + } +} diff --git a/filter.go b/filter.go index 94c18af..2bb8379 100644 --- a/filter.go +++ b/filter.go @@ -80,12 +80,14 @@ func EvaluateFilter(record map[string]interface{}, query map[string]interface{}) _, exists := record[dim] return exists } + return false case "selector": dim, dimOk := query["dimension"].(string) val, valOk := query["value"] if dimOk && valOk { return record[dim] == val } + return false case "and": if fields, ok := query["fields"].([]interface{}); ok { for _, f := range fields { diff --git a/filter_test.go b/filter_test.go new file mode 100644 index 0000000..06fb394 --- /dev/null +++ b/filter_test.go @@ -0,0 +1,424 @@ +package main + +import ( + "os" + "path/filepath" + "testing" +) + +func TestEvaluateFilter_EmptyOrMissingType(t *testing.T) { + record := map[string]interface{}{"command": "SELECT", "dbUserName": "app_user"} + + // Empty query matches all + if !EvaluateFilter(record, map[string]interface{}{}) { + t.Errorf("expected empty query to return true") + } + + // Missing type field matches all + if !EvaluateFilter(record, map[string]interface{}{"foo": "bar"}) { + t.Errorf("expected query without type to return true") + } + + // Unknown type matches all + if !EvaluateFilter(record, map[string]interface{}{"type": "unknownType"}) { + t.Errorf("expected unknown type to return true") + } +} + +func TestEvaluateFilter_DimensionExists(t *testing.T) { + record := map[string]interface{}{ + "command": "SELECT", + "dbUserName": "alice", + } + + // Dimension exists + queryMatch := map[string]interface{}{ + "type": "dimensionExists", + "dimension": "command", + } + if !EvaluateFilter(record, queryMatch) { + t.Errorf("expected dimensionExists to return true for existing dimension") + } + + // Dimension does not exist + queryNoMatch := map[string]interface{}{ + "type": "dimensionExists", + "dimension": "nonExistent", + } + if EvaluateFilter(record, queryNoMatch) { + t.Errorf("expected dimensionExists to return false for missing dimension") + } + + // Invalid dimension type (not string) + queryInvalid := map[string]interface{}{ + "type": "dimensionExists", + "dimension": 12345, + } + if EvaluateFilter(record, queryInvalid) { + t.Errorf("expected dimensionExists to return false for non-string dimension") + } +} + +func TestEvaluateFilter_Selector(t *testing.T) { + record := map[string]interface{}{ + "command": "SELECT", + "dbUserName": "alice", + "rowCount": float64(10), + } + + // Match string + matchQuery := map[string]interface{}{ + "type": "selector", + "dimension": "command", + "value": "SELECT", + } + if !EvaluateFilter(record, matchQuery) { + t.Errorf("expected selector to match string value") + } + + // Non-match string + noMatchQuery := map[string]interface{}{ + "type": "selector", + "dimension": "command", + "value": "DROP", + } + if EvaluateFilter(record, noMatchQuery) { + t.Errorf("expected selector to not match different value") + } + + // Non-existent dimension + missingDimQuery := map[string]interface{}{ + "type": "selector", + "dimension": "missingField", + "value": "val", + } + if EvaluateFilter(record, missingDimQuery) { + t.Errorf("expected selector to not match missing dimension") + } + + // Match number + matchNumQuery := map[string]interface{}{ + "type": "selector", + "dimension": "rowCount", + "value": float64(10), + } + if !EvaluateFilter(record, matchNumQuery) { + t.Errorf("expected selector to match numeric value") + } +} + +func TestEvaluateFilter_Not(t *testing.T) { + record := map[string]interface{}{ + "command": "SELECT", + } + + // NOT match (inner matches -> NOT returns false) + notMatchQuery := map[string]interface{}{ + "type": "not", + "field": map[string]interface{}{ + "type": "selector", + "dimension": "command", + "value": "SELECT", + }, + } + if EvaluateFilter(record, notMatchQuery) { + t.Errorf("expected not(selector=SELECT) to return false when command is SELECT") + } + + // NOT non-match (inner doesn't match -> NOT returns true) + notNoMatchQuery := map[string]interface{}{ + "type": "not", + "field": map[string]interface{}{ + "type": "selector", + "dimension": "command", + "value": "INSERT", + }, + } + if !EvaluateFilter(record, notNoMatchQuery) { + t.Errorf("expected not(selector=INSERT) to return true when command is SELECT") + } + + // Malformed field + notMalformed := map[string]interface{}{ + "type": "not", + "field": "invalid-not-map", + } + if !EvaluateFilter(record, notMalformed) { + t.Errorf("expected malformed not to return default true") + } +} + +func TestEvaluateFilter_And(t *testing.T) { + record := map[string]interface{}{ + "command": "SELECT", + "dbUserName": "alice", + } + + // Both match + andAllMatch := map[string]interface{}{ + "type": "and", + "fields": []interface{}{ + map[string]interface{}{"type": "selector", "dimension": "command", "value": "SELECT"}, + map[string]interface{}{"type": "selector", "dimension": "dbUserName", "value": "alice"}, + }, + } + if !EvaluateFilter(record, andAllMatch) { + t.Errorf("expected and query to return true when all match") + } + + // One fails + andOneFails := map[string]interface{}{ + "type": "and", + "fields": []interface{}{ + map[string]interface{}{"type": "selector", "dimension": "command", "value": "SELECT"}, + map[string]interface{}{"type": "selector", "dimension": "dbUserName", "value": "bob"}, + }, + } + if EvaluateFilter(record, andOneFails) { + t.Errorf("expected and query to return false when one subquery fails") + } + + // Non-slice fields + andMalformed := map[string]interface{}{ + "type": "and", + "fields": "not-a-slice", + } + if !EvaluateFilter(record, andMalformed) { + t.Errorf("expected malformed and query to return true") + } +} + +func TestEvaluateFilter_Or(t *testing.T) { + record := map[string]interface{}{ + "command": "SELECT", + "dbUserName": "alice", + } + + // One matches + orOneMatches := map[string]interface{}{ + "type": "or", + "fields": []interface{}{ + map[string]interface{}{"type": "selector", "dimension": "command", "value": "DROP"}, + map[string]interface{}{"type": "selector", "dimension": "dbUserName", "value": "alice"}, + }, + } + if !EvaluateFilter(record, orOneMatches) { + t.Errorf("expected or query to return true when at least one matches") + } + + // None match + orNoneMatch := map[string]interface{}{ + "type": "or", + "fields": []interface{}{ + map[string]interface{}{"type": "selector", "dimension": "command", "value": "DROP"}, + map[string]interface{}{"type": "selector", "dimension": "dbUserName", "value": "bob"}, + }, + } + if EvaluateFilter(record, orNoneMatch) { + t.Errorf("expected or query to return false when none match") + } + + // Malformed fields + orMalformed := map[string]interface{}{ + "type": "or", + "fields": "invalid", + } + if EvaluateFilter(record, orMalformed) { + t.Errorf("expected malformed or query to return false") + } +} + +func TestEvaluateFilter_Contains(t *testing.T) { + record := map[string]interface{}{ + "stringField": "hello world from database", + "sliceField": []interface{}{"item1", "item2", "item3"}, + "mapField": map[string]interface{}{ + "keyA": "valueA", + "keyB": "valueB", + }, + "numField": 42, + } + + // String substring match + strContains := map[string]interface{}{ + "type": "contains", + "dimension": "stringField", + "value": "world", + } + if !EvaluateFilter(record, strContains) { + t.Errorf("expected string contains to match") + } + + // String substring non-match + strNoContains := map[string]interface{}{ + "type": "contains", + "dimension": "stringField", + "value": "notfound", + } + if EvaluateFilter(record, strNoContains) { + t.Errorf("expected string contains to not match") + } + + // Slice item match + sliceContains := map[string]interface{}{ + "type": "contains", + "dimension": "sliceField", + "value": "item2", + } + if !EvaluateFilter(record, sliceContains) { + t.Errorf("expected slice contains to match item") + } + + // Slice item non-match + sliceNoContains := map[string]interface{}{ + "type": "contains", + "dimension": "sliceField", + "value": "item99", + } + if EvaluateFilter(record, sliceNoContains) { + t.Errorf("expected slice contains to not match missing item") + } + + // Map key match + mapContains := map[string]interface{}{ + "type": "contains", + "dimension": "mapField", + "value": "keyA", + } + if !EvaluateFilter(record, mapContains) { + t.Errorf("expected map contains to match key") + } + + // Map key non-match + mapNoContains := map[string]interface{}{ + "type": "contains", + "dimension": "mapField", + "value": "keyZ", + } + if EvaluateFilter(record, mapNoContains) { + t.Errorf("expected map contains to not match missing key") + } + + // Missing dimension + missingDim := map[string]interface{}{ + "type": "contains", + "dimension": "missingField", + "value": "val", + } + if EvaluateFilter(record, missingDim) { + t.Errorf("expected contains to return false for missing dimension") + } + + // Unsupported type (number) + unsupportedType := map[string]interface{}{ + "type": "contains", + "dimension": "numField", + "value": 42, + } + if EvaluateFilter(record, unsupportedType) { + t.Errorf("expected contains to return false for unsupported type") + } +} + +func TestLoadFilterConfig(t *testing.T) { + // Create temp directory for testing filters + tmpDir := t.TempDir() + origWd, err := os.Getwd() + if err != nil { + t.Fatalf("failed to get wd: %v", err) + } + defer func() { + _ = os.Chdir(origWd) + }() + + err = os.Chdir(tmpDir) + if err != nil { + t.Fatalf("failed to chdir to temp dir: %v", err) + } + + // 1. Missing filter and no default returns empty config + cfg, err := LoadFilterConfig("nonexistent") + if err != nil { + t.Fatalf("expected nil error on missing filter, got %v", err) + } + if len(cfg.Drop) != 0 || len(cfg.Query) != 0 { + t.Errorf("expected empty config on missing filter") + } + + // 2. Load JSON filter + filtersDir := filepath.Join(tmpDir, "filters") + if err := os.MkdirAll(filtersDir, 0755); err != nil { + t.Fatalf("failed to mkdir filters: %v", err) + } + + jsonContent := []byte(`{ + "drop": ["paramList", "errorMessage"], + "query": {"type": "selector", "dimension": "command", "value": "SELECT"} + }`) + if err := os.WriteFile(filepath.Join(filtersDir, "custom.json"), jsonContent, 0644); err != nil { + t.Fatalf("failed to write custom.json: %v", err) + } + + cfg, err = LoadFilterConfig("custom") + if err != nil { + t.Fatalf("expected nil error loading JSON filter, got: %v", err) + } + if len(cfg.Drop) != 2 || cfg.Drop[0] != "paramList" { + t.Errorf("unexpected Drop fields: %+v", cfg.Drop) + } + if cfg.Query["type"] != "selector" { + t.Errorf("unexpected Query type: %v", cfg.Query["type"]) + } + + // 3. Load YAML filter + yamlContent := []byte(` +drop: + - clientApplication +query: + type: dimensionExists + dimension: objectName +`) + if err := os.WriteFile(filepath.Join(filtersDir, "audit.yaml"), yamlContent, 0644); err != nil { + t.Fatalf("failed to write audit.yaml: %v", err) + } + + cfg, err = LoadFilterConfig("audit") + if err != nil { + t.Fatalf("expected nil error loading YAML filter, got: %v", err) + } + if len(cfg.Drop) != 1 || cfg.Drop[0] != "clientApplication" { + t.Errorf("unexpected Drop fields from YAML: %+v", cfg.Drop) + } + + // 4. Fallback to default.json + defaultContent := []byte(`{"drop": ["dropFromDefault"], "query": {}}`) + if err := os.WriteFile(filepath.Join(filtersDir, "default.json"), defaultContent, 0644); err != nil { + t.Fatalf("failed to write default.json: %v", err) + } + + cfg, err = LoadFilterConfig("missing-filter-name") + if err != nil { + t.Fatalf("expected nil error falling back to default, got: %v", err) + } + if len(cfg.Drop) != 1 || cfg.Drop[0] != "dropFromDefault" { + t.Errorf("expected fallback to default.json, got %+v", cfg.Drop) + } + + // 5. Malformed file returns error + if err := os.WriteFile(filepath.Join(filtersDir, "malformed.json"), []byte("{bad json"), 0644); err != nil { + t.Fatalf("failed to write malformed.json: %v", err) + } + _, err = LoadFilterConfig("malformed") + if err == nil { + t.Errorf("expected error loading malformed json, got nil") + } + + // 6. Malformed YAML returns error + if err := os.WriteFile(filepath.Join(filtersDir, "malformed.yaml"), []byte(":\n - : bad yaml"), 0644); err != nil { + t.Fatalf("failed to write malformed.yaml: %v", err) + } + _, err = LoadFilterConfig("malformed") + if err == nil { + t.Errorf("expected error loading malformed yaml, got nil") + } +} diff --git a/main.go b/main.go index af6d70b..5bcef3f 100644 --- a/main.go +++ b/main.go @@ -255,7 +255,10 @@ func decryptAWSEncryptionSDKPayload(ctx context.Context, ciphertext, plaintextDa return decryptOutput.Plaintext, nil } -// decompressZlib decompresses zlib formatted data +// maxDecompressedSize limits decompression to 256MB to mitigate decompression bombs. +const maxDecompressedSize = 256 * 1024 * 1024 + +// decompressZlib decompresses zlib formatted data with decompression bomb protection. func decompressZlib(data []byte) ([]byte, error) { reader, err := zlib.NewReader(bytes.NewReader(data)) if err != nil { @@ -264,7 +267,7 @@ func decompressZlib(data []byte) ([]byte, error) { defer reader.Close() var out bytes.Buffer - if _, err := io.Copy(&out, reader); err != nil { + if _, err := io.Copy(&out, io.LimitReader(reader, maxDecompressedSize)); err != nil { return nil, err } return out.Bytes(), nil @@ -305,7 +308,7 @@ func convertJSONToParquet(decompressedJSON []byte, filterConfig *FilterConfig) ( } var buf bytes.Buffer - writer := parquet.NewWriter(&buf, parquet.Compression(&snappy.Codec{})) + writer := parquet.NewWriter(&buf, parquet.SchemaOf(new(DatabaseActivityEvent)), parquet.Compression(&snappy.Codec{})) for _, eventMap := range container.DatabaseActivityEventList { if t, ok := eventMap["type"].(string); ok && t == "heartbeat" { diff --git a/testdata/das_empty.json b/testdata/das_empty.json new file mode 100644 index 0000000..b52712e --- /dev/null +++ b/testdata/das_empty.json @@ -0,0 +1,3 @@ +{ + "databaseActivityEventList": [] +} diff --git a/testdata/das_events.json b/testdata/das_events.json new file mode 100644 index 0000000..f3f1c07 --- /dev/null +++ b/testdata/das_events.json @@ -0,0 +1,53 @@ +{ + "databaseActivityEventList": [ + { + "type": "heartbeat" + }, + { + "type": "activity", + "logTime": "2026-09-29 12:00:00.000", + "statementId": 1001, + "substatementId": 0, + "objectType": "TABLE", + "command": "SELECT", + "objectName": "users", + "databaseName": "production_db", + "dbUserName": "app_user", + "remoteHost": "10.0.1.45", + "sessionId": "sess-998877", + "rowCount": 42, + "commandText": "SELECT id, name, email FROM users WHERE active = true", + "paramList": ["true"], + "pid": 12345, + "clientApplication": "backend-service", + "exitCode": "0", + "class": "READ", + "serverHost": "ip-10-0-2-10", + "startTime": "2026-09-29 12:00:00.000", + "errorMessage": "" + }, + { + "type": "activity", + "logTime": "2026-09-29 12:00:01.000", + "statementId": 1002, + "substatementId": 0, + "objectType": "TABLE", + "command": "UPDATE", + "objectName": "users", + "databaseName": "production_db", + "dbUserName": "admin", + "remoteHost": "10.0.1.50", + "sessionId": "sess-112233", + "rowCount": 1, + "commandText": "UPDATE users SET active = false WHERE id = 123", + "paramList": ["false", "123"], + "pid": 12346, + "clientApplication": "admin-portal", + "exitCode": "0", + "class": "WRITE", + "serverHost": "ip-10-0-2-10", + "startTime": "2026-09-29 12:00:01.000", + "errorMessage": "" + } + ] +} diff --git a/testdata/das_heartbeat_only.json b/testdata/das_heartbeat_only.json new file mode 100644 index 0000000..dea1459 --- /dev/null +++ b/testdata/das_heartbeat_only.json @@ -0,0 +1,10 @@ +{ + "databaseActivityEventList": [ + { + "type": "heartbeat" + }, + { + "type": "heartbeat" + } + ] +} diff --git a/testdata/eventbridge_s3_event.json b/testdata/eventbridge_s3_event.json new file mode 100644 index 0000000..a93d822 --- /dev/null +++ b/testdata/eventbridge_s3_event.json @@ -0,0 +1,26 @@ +{ + "version": "0", + "id": "eventbridge-id-98765", + "detail-type": "Object Created", + "source": "aws.s3", + "account": "123456789012", + "time": "2026-09-29T12:00:00Z", + "region": "us-east-1", + "resources": [ + "arn:aws:s3:::aws-das-source-bucket" + ], + "detail": { + "version": "0", + "bucket": { + "name": "aws-das-source-bucket" + }, + "object": { + "key": "AWSLogs/123456789012/RDS/us-east-1/cluster-ABCD/2026/09/29/das-data.json.gz", + "size": 1024, + "etag": "d41d8cd98f00b204e9800998ecf8427e", + "sequencer": "0055AED6DCD90281E5" + }, + "request-id": "C3D13FE58DE4C810", + "requester": "123456789012" + } +} diff --git a/testdata/s3_event_notification.json b/testdata/s3_event_notification.json new file mode 100644 index 0000000..ad17215 --- /dev/null +++ b/testdata/s3_event_notification.json @@ -0,0 +1,27 @@ +{ + "Records": [ + { + "eventVersion": "2.1", + "eventSource": "aws:s3", + "awsRegion": "us-east-1", + "eventTime": "2026-09-29T12:00:00.000Z", + "eventName": "ObjectCreated:Put", + "userIdentity": { + "principalId": "AWS:AIDAIEXAMPLE" + }, + "s3": { + "s3SchemaVersion": "1.0", + "configurationId": "das-event-notification", + "bucket": { + "name": "aws-das-source-bucket", + "arn": "arn:aws:s3:::aws-das-source-bucket" + }, + "object": { + "key": "AWSLogs/123456789012/RDS/us-east-1/cluster-ABCD/2026/09/29/das-data.json.gz", + "size": 1024, + "eTag": "d41d8cd98f00b204e9800998ecf8427e" + } + } + } + ] +} diff --git a/testdata/sns_s3_event.json b/testdata/sns_s3_event.json new file mode 100644 index 0000000..15c00e5 --- /dev/null +++ b/testdata/sns_s3_event.json @@ -0,0 +1,8 @@ +{ + "Type": "Notification", + "MessageId": "sns-message-id-12345", + "TopicArn": "arn:aws:sns:us-east-1:123456789012:das-fanout-topic", + "Subject": "Amazon S3 Notification", + "Message": "{\"Records\":[{\"eventVersion\":\"2.1\",\"eventSource\":\"aws:s3\",\"awsRegion\":\"us-east-1\",\"eventName\":\"ObjectCreated:Put\",\"s3\":{\"bucket\":{\"name\":\"aws-das-source-bucket\"},\"object\":{\"key\":\"AWSLogs/123456789012/RDS/us-east-1/cluster-ABCD/2026/09/29/das-data.json.gz\"}}}]}", + "Timestamp": "2026-09-29T12:00:01.000Z" +}