diff --git a/pkg/database/alerts.go b/pkg/database/alerts.go index 0ba0dc6e460..86f2e8c9059 100644 --- a/pkg/database/alerts.go +++ b/pkg/database/alerts.go @@ -436,109 +436,123 @@ func parseAlertTimes(alert *models.Alert, logger log.FieldLogger) (time.Time, ti return start, stop } -func buildEventCreates(ctx context.Context, logger log.FieldLogger, client *ent.Client, machineID string, alertItem *models.Alert) ([]*ent.Event, error) { - // let's track when we strip or drop data, notify outside of loop to avoid spam +func marshalEventSerialized(logger log.FieldLogger, machineID, scenario string, eventItem *models.Event) (string, bool, bool, error) { stripped := false dropped := false - if len(alertItem.Events) == 0 { - return nil, nil + marshallMetas, err := json.Marshal(eventItem.Meta) + if err != nil { + return "", stripped, dropped, fmt.Errorf("event meta '%v': %w: %w", eventItem.Meta, err, MarshalFail) } - eventBulk := make([]*ent.EventCreate, len(alertItem.Events)) - - for i, eventItem := range alertItem.Events { - ts, err := time.Parse(time.RFC3339, *eventItem.Timestamp) - if err != nil { - logger.Errorf("creating alert: Failed to parse event timestamp '%s', defaulting to now: %s", *eventItem.Timestamp, err) - - ts = time.Now().UTC() - } + // the serialized field is too big, let's try to progressively strip it + if event.SerializedValidator(string(marshallMetas)) != nil { + stripped = true - marshallMetas, err := json.Marshal(eventItem.Meta) - if err != nil { - return nil, fmt.Errorf("event meta '%v': %w: %w", eventItem.Meta, err, MarshalFail) - } - - // the serialized field is too big, let's try to progressively strip it - if event.SerializedValidator(string(marshallMetas)) != nil { - stripped = true - - valid := false - stripSize := 2048 - - for !valid && stripSize > 0 { - for _, serializedItem := range eventItem.Meta { - if len(serializedItem.Value) > stripSize*2 { - serializedItem.Value = serializedItem.Value[:stripSize] + "" - } - } - - marshallMetas, err = json.Marshal(eventItem.Meta) - if err != nil { - return nil, fmt.Errorf("event meta '%v': %w: %w", eventItem.Meta, err, MarshalFail) - } + valid := false + stripSize := 2048 - if event.SerializedValidator(string(marshallMetas)) == nil { - valid = true + for !valid && stripSize > 0 { + for _, serializedItem := range eventItem.Meta { + if len(serializedItem.Value) > stripSize*2 { + serializedItem.Value = serializedItem.Value[:stripSize] + "" } + } - stripSize /= 2 + marshallMetas, err = json.Marshal(eventItem.Meta) + if err != nil { + return "", stripped, dropped, fmt.Errorf("event meta '%v': %w: %w", eventItem.Meta, err, MarshalFail) } - // nothing worked, drop it - if !valid { - dropped = true - stripped = false - marshallMetas = []byte("") + if event.SerializedValidator(string(marshallMetas)) == nil { + valid = true } + + stripSize /= 2 } - eventBulk[i] = client.Event.Create(). - SetTime(ts). - SetSerialized(string(marshallMetas)) + // nothing worked, drop it + if !valid { + dropped = true + stripped = false + marshallMetas = []byte("") + } } if stripped { - logger.Warningf("stripped 'serialized' field (machine %s / scenario %s)", machineID, *alertItem.Scenario) + logger.Warningf("stripped 'serialized' field (machine %s / scenario %s)", machineID, scenario) } if dropped { - logger.Warningf("dropped 'serialized' field (machine %s / scenario %s)", machineID, *alertItem.Scenario) + logger.Warningf("dropped 'serialized' field (machine %s / scenario %s)", machineID, scenario) } - return client.Event.CreateBulk(eventBulk...).Save(ctx) + return string(marshallMetas), stripped, dropped, nil } -func buildMetaCreates(ctx context.Context, logger log.FieldLogger, client *ent.Client, alertItem *models.Alert) ([]*ent.Meta, error) { - if len(alertItem.Meta) == 0 { - return nil, nil +// createAlertEventsForAlert inserts events with owner set at creation time. +// Pre-creating events then linking via AddEvents fails on MySQL (alert_events constraint). +func createAlertEventsForAlert(ctx context.Context, logger log.FieldLogger, client *ent.Client, machineID, scenario string, alertID int, events []*models.Event) error { + for _, eventItem := range events { + ts, err := time.Parse(time.RFC3339, *eventItem.Timestamp) + if err != nil { + logger.Errorf("creating alert: Failed to parse event timestamp '%s', defaulting to now: %s", *eventItem.Timestamp, err) + + ts = time.Now().UTC() + } + + serialized, _, _, err := marshalEventSerialized(logger, machineID, scenario, eventItem) + if err != nil { + return err + } + + if _, err := client.Event.Create(). + SetTime(ts). + SetSerialized(serialized). + SetOwnerID(alertID). + Save(ctx); err != nil { + return fmt.Errorf("creating alert event: %w", err) + } } - metaBulk := make([]*ent.MetaCreate, len(alertItem.Meta)) + return nil +} - for i, metaItem := range alertItem.Meta { - key := metaItem.Key - value := metaItem.Value +func normalizeMetaItem(logger log.FieldLogger, metaItem *models.MetaItems0) (string, string) { + key := metaItem.Key + value := metaItem.Value - if len(metaItem.Value) > 4095 { - logger.Warningf("truncated meta %s: value too long", metaItem.Key) + if len(metaItem.Value) > 4095 { + logger.Warningf("truncated meta %s: value too long", metaItem.Key) - value = value[:4095] - } + value = value[:4095] + } - if len(metaItem.Key) > 255 { - logger.Warningf("truncated meta %s: key too long", metaItem.Key) + if len(metaItem.Key) > 255 { + logger.Warningf("truncated meta %s: key too long", metaItem.Key) - key = key[:255] - } + key = key[:255] + } + + return key, value +} + +// createAlertMetasForAlert inserts alert-level metas with owner set at creation time. +// Pre-creating metas then linking via AddMetas fails on MySQL (alert_metas constraint). +func createAlertMetasForAlert(ctx context.Context, logger log.FieldLogger, client *ent.Client, alertID int, metaItems []*models.MetaItems0) error { + for _, metaItem := range metaItems { + key, value := normalizeMetaItem(logger, metaItem) - metaBulk[i] = client.Meta.Create(). + if _, err := client.Meta.Create(). SetKey(key). - SetValue(value) + SetValue(value). + SetOwnerID(alertID). + Save(ctx); err != nil { + return fmt.Errorf("creating alert meta %s: %w", key, err) + } } - return client.Meta.CreateBulk(metaBulk...).Save(ctx) + return nil } func (c *Client) buildDecisions(ctx context.Context, logger log.FieldLogger, client *ent.Client, alertItem *models.Alert, stopAtTime time.Time) ([]*ent.Decision, int, error) { @@ -588,25 +602,32 @@ func (c *Client) saveAlerts(ctx context.Context, client *ent.Client, batch []ale return nil, nil } - // extract builders in the same order - builders := make([]*ent.AlertCreate, len(batch)) + // Insert alerts one-by-one; events and metas are created after save with owner set. + ret := make([]string, len(batch)) for i := range batch { if batch[i].builder == nil { return nil, fmt.Errorf("nil alert builder at index %d", i) } - builders[i] = batch[i].builder - } - - alertsCreateBulk, err := client.Alert.CreateBulk(builders...).Save(ctx) - if err != nil { - return nil, fmt.Errorf("bulk creating alert: %w: %w", err, BulkError) - } + a, err := batch[i].builder.Save(ctx) + if err != nil { + return nil, fmt.Errorf("creating alert: %w: %w", err, BulkError) + } - ret := make([]string, len(alertsCreateBulk)) - for i, a := range alertsCreateBulk { ret[i] = strconv.Itoa(a.ID) + if len(batch[i].events) > 0 { + if err := createAlertEventsForAlert(ctx, c.Log, client, batch[i].machineID, batch[i].scenario, a.ID, batch[i].events); err != nil { + return nil, fmt.Errorf("creating events for alert %d: %w: %w", a.ID, err, BulkError) + } + } + + if len(batch[i].meta) > 0 { + if err := createAlertMetasForAlert(ctx, c.Log, client, a.ID, batch[i].meta); err != nil { + return nil, fmt.Errorf("creating metas for alert %d: %w: %w", a.ID, err, BulkError) + } + } + d := batch[i].decisions if len(d) == 0 { continue @@ -627,6 +648,10 @@ func (c *Client) saveAlerts(ctx context.Context, client *ent.Client, batch []ale type alertCreatePlan struct { builder *ent.AlertCreate + machineID string + scenario string + events []*models.Event + meta []*models.MetaItems0 decisions []*ent.Decision } @@ -648,16 +673,6 @@ func (c *Client) createAlertBatch(ctx context.Context, machineID string, owner * c.Log.Info(disp) } - events, err := buildEventCreates(ctx, c.Log, txEnt, machineID, alertItem) - if err != nil { - return nil, rollbackOnError(tx, err, fmt.Sprintf("building events for alert %s", alertItem.UUID)) - } - - metas, err := buildMetaCreates(ctx, c.Log, txEnt, alertItem) - if err != nil { - c.Log.Warningf("error creating alert meta: %s", err) - } - decisions, discardCount, err := c.buildDecisions(ctx, c.Log, txEnt, alertItem, stopAtTime) if err != nil { return nil, rollbackOnError(tx, err, fmt.Sprintf("building decisions for alert %s", alertItem.UUID)) @@ -692,9 +707,7 @@ func (c *Client) createAlertBatch(ctx context.Context, machineID string, owner * SetScenarioHash(*alertItem.ScenarioHash). SetRemediation(alertItem.Remediation). SetUUID(alertItem.UUID). - SetKind(alertItem.Kind). - AddEvents(events...). - AddMetas(metas...) + SetKind(alertItem.Kind) if owner != nil { builder.SetOwner(owner) @@ -702,6 +715,10 @@ func (c *Client) createAlertBatch(ctx context.Context, machineID string, owner * batch = append(batch, alertCreatePlan{ builder: builder, + machineID: machineID, + scenario: *alertItem.Scenario, + events: alertItem.Events, + meta: alertItem.Meta, decisions: decisions, }) } diff --git a/pkg/database/alerts_appsec_metas_test.go b/pkg/database/alerts_appsec_metas_test.go new file mode 100644 index 00000000000..ffb6a53569d --- /dev/null +++ b/pkg/database/alerts_appsec_metas_test.go @@ -0,0 +1,205 @@ +package database + +import ( + "context" + "database/sql" + "fmt" + "os" + "testing" + "time" + + _ "github.com/go-sql-driver/mysql" + + "github.com/go-openapi/strfmt" + "github.com/google/uuid" + "github.com/stretchr/testify/require" + + "github.com/crowdsecurity/crowdsec/pkg/csconfig" + "github.com/crowdsecurity/crowdsec/pkg/models" + "github.com/crowdsecurity/crowdsec/pkg/types" +) + +// Set CROWDSEC_MYSQL_APPSEC_ALERT_TEST=1 to run against MySQL on 127.0.0.1:3306. +// Exercises the AppSec alert insert path: multiple events + alert-level metas, kind=waf. +const mysqlAppsecAlertTestEnv = "CROWDSEC_MYSQL_APPSEC_ALERT_TEST" + +const ( + mysqlMetasHost = "127.0.0.1" + mysqlMetasPort = 3306 + mysqlMetasUser = "crowdsec" + mysqlMetasPassword = "crowdsec" +) + +func TestCreateAppsecLikeAlert_SQLite(t *testing.T) { + ctx := t.Context() + + dbClient, err := NewClient(ctx, &csconfig.DatabaseCfg{ + Type: "sqlite", + DbName: "crowdsec", + DbPath: ":memory:", + }, nil) + require.NoError(t, err) + t.Cleanup(func() { _ = dbClient.Close() }) + + runAppsecAlertInsertTest(t, ctx, dbClient) +} + +func TestCreateAppsecLikeAlert_MySQL(t *testing.T) { + if os.Getenv(mysqlAppsecAlertTestEnv) == "" { + t.Skipf("set %s=1 to run (requires mysql on %s:%d)", mysqlAppsecAlertTestEnv, mysqlMetasHost, mysqlMetasPort) + } + + ctx := t.Context() + dbName := fmt.Sprintf("crowdsec_metas_test_%d", time.Now().UnixNano()) + + user := envOrDefault("CROWDSEC_MYSQL_USER", mysqlMetasUser) + password := envOrDefault("CROWDSEC_MYSQL_PASSWORD", mysqlMetasPassword) + + createMySQLDatabase(t, ctx, dbName, user, password) + t.Cleanup(func() { dropMySQLDatabase(t, dbName, user, password) }) + + dbClient, err := NewClient(ctx, &csconfig.DatabaseCfg{ + Type: "mysql", + Host: mysqlMetasHost, + Port: mysqlMetasPort, + User: user, + Password: password, + DbName: dbName, + }, nil) + require.NoError(t, err) + t.Cleanup(func() { _ = dbClient.Close() }) + + runAppsecAlertInsertTest(t, ctx, dbClient) +} + +func runAppsecAlertInsertTest(t *testing.T, ctx context.Context, dbClient *Client) { + t.Helper() + + machineID := "appsec-metas-test" + pw := strfmt.Password("password") + _, err := dbClient.CreateMachine(ctx, &machineID, &pw, "127.0.0.1", true, true, types.PasswordAuthType) + require.NoError(t, err) + + // Single alert (typical AppSec POST) + ids, err := dbClient.CreateAlert(ctx, machineID, []*models.Alert{makeAppsecLikeAlert(0)}) + require.NoError(t, err, "single appsec-like alert insert failed") + require.Len(t, ids, 1) + + // Batch of distinct alerts (exercises former CreateBulk path) + batch := make([]*models.Alert, 5) + for i := range batch { + batch[i] = makeAppsecLikeAlert(i) + } + + ids, err = dbClient.CreateAlert(ctx, machineID, batch) + require.NoError(t, err, "batched appsec-like alert insert failed") + require.Len(t, ids, len(batch)) +} + +// makeAppsecLikeAlert mirrors the AppSec payload shape: multiple events, +// alert-level context metas (~7 keys), kind=waf, no decisions. +func makeAppsecLikeAlert(seed int) *models.Alert { + now := time.Now().UTC().Format(time.RFC3339) + + scenario := "crowdsecurity/vpatch-env-access" + scenarioVersion := "0.1" + scenarioHash := "deadbeef" + message := "AppSec detected a threat" + leakSpeed := "" + scope := "Ip" + value := fmt.Sprintf("185.157.63.%d", seed+18) + capacity := int32(1) + eventsCount := int32(3) + simulated := false + kind := types.WAFAlertKind.String() + remediation := true + + events := make([]*models.Event, 3) + for i := range events { + ts := now + events[i] = &models.Event{ + Timestamp: &ts, + Meta: models.Meta{ + {Key: "rule_name", Value: scenario}, + {Key: "uri", Value: "/.env"}, + {Key: "message", Value: "Env file access"}, + }, + } + } + + alertMeta := models.Meta{ + {Key: "method", Value: "GET"}, + {Key: "uri", Value: "/.env"}, + {Key: "rules", Value: "crowdsecurity/vpatch-env-access"}, + {Key: "id", Value: fmt.Sprintf("rule-%d", seed)}, + {Key: "name", Value: scenario}, + {Key: "target_fqdn", Value: "test.example.com"}, + {Key: "matched_zones", Value: "REQUEST_URI"}, + } + + return &models.Alert{ + UUID: uuid.NewString(), + Capacity: &capacity, + Scenario: &scenario, + ScenarioVersion: &scenarioVersion, + ScenarioHash: &scenarioHash, + Message: &message, + Leakspeed: &leakSpeed, + EventsCount: &eventsCount, + Simulated: &simulated, + StartAt: &now, + StopAt: &now, + Kind: kind, + Remediation: remediation, + Source: &models.Source{ + Scope: &scope, + Value: &value, + IP: value, + }, + Events: events, + Meta: alertMeta, + } +} + +func envOrDefault(key, fallback string) string { + if v := os.Getenv(key); v != "" { + return v + } + + return fallback +} + +func createMySQLDatabase(t *testing.T, ctx context.Context, dbName, user, password string) { + t.Helper() + + adminDSN := fmt.Sprintf("%s:%s@tcp(%s:%d)/", user, password, mysqlMetasHost, mysqlMetasPort) + + adminDB, err := sql.Open("mysql", adminDSN) + require.NoError(t, err, "connecting to admin mysql") + defer adminDB.Close() + + _, err = adminDB.ExecContext(ctx, fmt.Sprintf("DROP DATABASE IF EXISTS `%s`", dbName)) + require.NoError(t, err) + _, err = adminDB.ExecContext(ctx, fmt.Sprintf("CREATE DATABASE `%s`", dbName)) + require.NoError(t, err) +} + +func dropMySQLDatabase(t *testing.T, dbName, user, password string) { + t.Helper() + + adminDSN := fmt.Sprintf("%s:%s@tcp(%s:%d)/", user, password, mysqlMetasHost, mysqlMetasPort) + + adminDB, err := sql.Open("mysql", adminDSN) + if err != nil { + t.Logf("cleanup: open admin db: %s", err) + return + } + defer adminDB.Close() + + cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) //nolint:usetesting + defer cancel() + + if _, err := adminDB.ExecContext(cleanupCtx, fmt.Sprintf("DROP DATABASE IF EXISTS `%s`", dbName)); err != nil { + t.Logf("cleanup: drop db %s: %s", dbName, err) + } +}