diff --git a/internal/api/handler_accounts.go b/internal/api/handler_accounts.go index f62581362..bec9a02ce 100644 --- a/internal/api/handler_accounts.go +++ b/internal/api/handler_accounts.go @@ -21,6 +21,7 @@ import ( "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/service/sts" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" ) @@ -1173,7 +1174,11 @@ func (h *Handler) saveAccountServiceOverride(ctx context.Context, httpReq *event return nil, err } - if err := h.config.SaveAccountServiceOverride(ctx, override); err != nil { + // The write and its config_audit_events row commit together (issue #1493). + audit := config.AuditContext{Actor: configAuditActor(session)} + if err := h.config.WithTx(ctx, func(tx pgx.Tx) error { + return h.config.SaveAccountServiceOverrideAuditedTx(ctx, tx, override, audit) + }); err != nil { return nil, fmt.Errorf("accounts: %w", err) } @@ -1279,7 +1284,10 @@ func (h *Handler) deleteAccountServiceOverride(ctx context.Context, req *events. return nil, err } - if err := h.config.DeleteAccountServiceOverride(ctx, accountID, provider, service); err != nil { + audit := config.AuditContext{Actor: configAuditActor(session)} + if err := h.config.WithTx(ctx, func(tx pgx.Tx) error { + return h.config.DeleteAccountServiceOverrideAuditedTx(ctx, tx, accountID, provider, service, audit) + }); err != nil { return nil, fmt.Errorf("accounts: %w", err) } diff --git a/internal/api/handler_override_audit_integration_test.go b/internal/api/handler_override_audit_integration_test.go new file mode 100644 index 000000000..1d2a6aa0c --- /dev/null +++ b/internal/api/handler_override_audit_integration_test.go @@ -0,0 +1,246 @@ +//go:build integration + +package api + +// Real-Postgres tests for the account service override audit trail (issue +// #1493): every override write through the single PUT, the single DELETE and +// an applied import leaves config_audit_events rows with the before and after +// values, in the same transaction as the write. + +import ( + "context" + "encoding/json" + "fmt" + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/auth" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" +) + +type overrideAuditFixture struct { + pg *testhelpers.PostgresContainer + store *config.PostgresStore + h *Handler + acct *config.CloudAccount +} + +func newOverrideAuditFixture(ctx context.Context, t *testing.T) *overrideAuditFixture { + t.Helper() + pg := testhelpers.RequireMigratedDatabase(ctx, t, "../database/postgres/migrations") + store := config.NewPostgresStore(pg.DB) + acct := &config.CloudAccount{Name: "audit fixture", Provider: "aws", ExternalID: "123456789012", Enabled: true} + require.NoError(t, store.CreateCloudAccount(ctx, acct)) + mockAuth := new(MockAuthService) + mockAuth.On("ValidateSession", ctx, "admin-token").Return(adminAccountSession(), nil) + mockAuth.grantPermissionsScoped([]auth.Permission{{Action: auth.ActionAdmin, Resource: auth.ResourceAll}}, nil) + return &overrideAuditFixture{pg: pg, store: store, h: &Handler{auth: mockAuth, config: store}, acct: acct} +} + +// auditRow is one config_audit_events row, with before/after/details decoded. +type auditRow struct { + ID, EventType, Actor string + ParentID, ActorUserID *string + AccountID *string + Provider, Service *string + Before, After, Details map[string]any + BeforeNull, AfterNull bool +} + +func (f *overrideAuditFixture) events(ctx context.Context, t *testing.T) []auditRow { + t.Helper() + rows, err := f.pg.DB.Pool().Query(ctx, ` + SELECT id::text, event_type, actor, parent_event_id::text, actor_user_id::text, + account_id::text, provider, service, before, after, details + FROM config_audit_events ORDER BY seq`) + require.NoError(t, err) + defer rows.Close() + var out []auditRow + for rows.Next() { + var r auditRow + var before, after, details []byte + require.NoError(t, rows.Scan(&r.ID, &r.EventType, &r.Actor, &r.ParentID, &r.ActorUserID, + &r.AccountID, &r.Provider, &r.Service, &before, &after, &details)) + r.BeforeNull, r.AfterNull = before == nil, after == nil + decodeAuditJSON(t, before, &r.Before) + decodeAuditJSON(t, after, &r.After) + decodeAuditJSON(t, details, &r.Details) + out = append(out, r) + } + require.NoError(t, rows.Err()) + return out +} + +func decodeAuditJSON(t *testing.T, raw []byte, dst *map[string]any) { + t.Helper() + if raw != nil { + require.NoError(t, json.Unmarshal(raw, dst)) + } +} + +func (f *overrideAuditFixture) put(ctx context.Context, t *testing.T, service, body string) { + t.Helper() + _, err := f.h.saveAccountServiceOverride(ctx, adminRequest(body), f.acct.ID+"/service-overrides/aws/"+service) + require.NoError(t, err) +} + +func (f *overrideAuditFixture) del(ctx context.Context, t *testing.T, service string) { + t.Helper() + _, err := f.h.deleteAccountServiceOverride(ctx, adminRequest(""), f.acct.ID+"/service-overrides/aws/"+service) + require.NoError(t, err) +} + +// TestOverrideAudit_SinglePutAndDelete: an insert, an update, a no-op save, a +// delete and a delete of nothing. Each change is one event naming the actor, +// the account, provider and service, with the stored row before and after. +func TestOverrideAudit_SinglePutAndDelete(t *testing.T) { + ctx := context.Background() + f := newOverrideAuditFixture(ctx, t) + + f.put(ctx, t, "rds", `{"term":1,"coverage":50}`) + f.put(ctx, t, "rds", `{"term":3,"coverage":75.5}`) + f.put(ctx, t, "rds", `{"term":3,"coverage":75.5}`) // no change: no event + f.del(ctx, t, "rds") + f.del(ctx, t, "rds") // nothing to delete: no event + + ev := f.events(ctx, t) + require.Len(t, ev, 3) + for _, e := range ev { + assert.Equal(t, "admin@example.com", e.Actor) + assert.Equal(t, adminAccountSession().UserID, *e.ActorUserID) + assert.Equal(t, f.acct.ID, *e.AccountID) + assert.Equal(t, []string{"aws", "rds"}, []string{*e.Provider, *e.Service}) + assert.Nil(t, e.ParentID) + } + + insert, update, del := ev[0], ev[1], ev[2] + assert.Equal(t, "account_service_override.upsert", insert.EventType) + assert.True(t, insert.BeforeNull, "an insert has no before") + assert.Equal(t, 1.0, insert.After["term"]) + assert.Equal(t, 50.0, insert.After["coverage"]) + + assert.Equal(t, "account_service_override.upsert", update.EventType) + assert.Equal(t, insert.After, update.Before, "the update's before is the row the insert wrote") + assert.Equal(t, 75.5, update.After["coverage"]) + assert.Equal(t, 3.0, update.After["term"]) + + assert.Equal(t, "account_service_override.delete", del.EventType) + assert.Equal(t, update.After, del.Before) + assert.True(t, del.AfterNull, "a delete has no after") +} + +// TestOverrideAudit_Import: an applied import records one import event with +// its mode and counts, and one linked event per written row. Unchanged rows, +// dry runs and rejected strict imports write no events. +func TestOverrideAudit_Import(t *testing.T) { + ctx := context.Background() + f := newOverrideAuditFixture(ctx, t) + f.put(ctx, t, "ec2", `{"term":1}`) + f.put(ctx, t, "rds", `{"term":3}`) + base := len(f.events(ctx, t)) + + rows := []OverrideTransferRow{ + {AccountID: f.acct.ID, Provider: "aws", Service: "ec2", Term: ptr(3)}, // update + {AccountID: f.acct.ID, Provider: "aws", Service: "rds", Term: ptr(3)}, // unchanged + {AccountID: f.acct.ID, Provider: "aws", Service: "elasticache", Term: ptr(1)}, // insert + {AccountID: f.acct.ID, Provider: "aws", Service: "opensearch", Term: ptr(7)}, // rejected + } + + _, err := runImport(t, f.h, OverrideImportRequest{Overrides: rows, Mode: overrideImportStrict}) + require.Error(t, err) + _, err = runImport(t, f.h, OverrideImportRequest{Overrides: rows, Mode: overrideImportBestEffort, DryRun: true}) + require.NoError(t, err) + require.Len(t, f.events(ctx, t), base, "a rejected strict import and a dry run write no events") + + report, err := runImport(t, f.h, OverrideImportRequest{Overrides: rows, Mode: overrideImportBestEffort}) + require.NoError(t, err) + require.True(t, report.Applied) + + ev := f.events(ctx, t)[base:] + require.Len(t, ev, 3, "one import event and one event per written row") + imp := ev[0] + assert.Equal(t, "account_service_override.import", imp.EventType) + assert.Equal(t, "admin@example.com", imp.Actor) + assert.Nil(t, imp.AccountID) + assert.Equal(t, "best_effort", imp.Details["mode"]) + assert.Equal(t, map[string]any{"insert": 1.0, "update": 1.0, "unchanged": 1.0, "rejected": 1.0}, imp.Details["summary"]) + + byService := map[string]auditRow{} + for _, e := range ev[1:] { + assert.Equal(t, "account_service_override.upsert", e.EventType) + require.NotNil(t, e.ParentID) + assert.Equal(t, imp.ID, *e.ParentID) + byService[*e.Service] = e + } + assert.Equal(t, 1.0, byService["ec2"].Before["term"]) + assert.Equal(t, 3.0, byService["ec2"].After["term"]) + assert.Equal(t, 1.0, byService["ec2"].Details["import_row"]) + assert.True(t, byService["elasticache"].BeforeNull) + assert.Equal(t, 1.0, byService["elasticache"].After["term"]) + assert.Equal(t, 3.0, byService["elasticache"].Details["import_row"]) + + // A write the database refuses part way rolls back the rows and every + // event, the import event included. + _, err = f.pg.DB.Pool().Exec(ctx, ` + CREATE FUNCTION reject_audit_svc() RETURNS trigger AS $$ + BEGIN RAISE EXCEPTION 'svc rejected'; END $$ LANGUAGE plpgsql; + CREATE TRIGGER reject_audit_svc BEFORE INSERT ON account_service_overrides + FOR EACH ROW WHEN (NEW.service = 'zz-last') EXECUTE FUNCTION reject_audit_svc();`) + require.NoError(t, err) + before := len(f.events(ctx, t)) + _, err = runImport(t, f.h, OverrideImportRequest{Overrides: []OverrideTransferRow{ + {AccountID: f.acct.ID, Provider: "aws", Service: "ec2", Term: ptr(1)}, + {AccountID: f.acct.ID, Provider: "aws", Service: "zz-last", Term: ptr(1)}, + }}) + require.Error(t, err) + require.Len(t, f.events(ctx, t), before, "a failed import leaves no events") + got, err := f.store.GetAccountServiceOverride(ctx, f.acct.ID, "aws", "ec2") + require.NoError(t, err) + assert.Equal(t, 3, *got.Term, "and no override writes") +} + +// TestOverrideAudit_ConcurrentFirstSavesChain: two PUTs creating the same +// override at once. The second writer replaces the first's row, so its +// event's before must be the first event's after; an audit that read the +// row before the other insert committed would record before = null for +// both and lose the first change from the trail. +func TestOverrideAudit_ConcurrentFirstSavesChain(t *testing.T) { + ctx := context.Background() + f := newOverrideAuditFixture(ctx, t) + + const rounds = 20 + broken := 0 + for round := 1; round <= rounds; round++ { + service := fmt.Sprintf("svc-%02d", round) + start := make(chan struct{}) + var wg sync.WaitGroup + for _, body := range []string{`{"term":1}`, `{"coverage":40}`} { + wg.Add(1) + go func(body string) { + defer wg.Done() + <-start + _, err := f.h.saveAccountServiceOverride(ctx, adminRequest(body), f.acct.ID+"/service-overrides/aws/"+service) + assert.NoError(t, err) + }(body) + } + close(start) + wg.Wait() + + var mine []auditRow + for _, e := range f.events(ctx, t) { + if e.Service != nil && *e.Service == service { + mine = append(mine, e) + } + } + require.Len(t, mine, 2, "round %d: one event per PUT", round) + if !mine[0].BeforeNull || mine[1].BeforeNull || fmt.Sprint(mine[1].Before) != fmt.Sprint(mine[0].After) { + broken++ + t.Logf("round %d: first before=%v after=%v, second before=%v", round, mine[0].Before, mine[0].After, mine[1].Before) + } + } + require.Zero(t, broken, "%d of %d rounds recorded an audit chain that skips a write", broken, rounds) +} diff --git a/internal/api/handler_override_audit_test.go b/internal/api/handler_override_audit_test.go new file mode 100644 index 000000000..94e91eef2 --- /dev/null +++ b/internal/api/handler_override_audit_test.go @@ -0,0 +1,172 @@ +package api + +import ( + "context" + "encoding/json" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// Unit tests of the override audit trail wiring (issue #1493). The store +// behavior (locks, before/after, no-op skip) is covered against Postgres in +// handler_override_audit_integration_test.go. + +func adminAuditActor() config.AuditActor { + uid := adminAccountSession().UserID + return config.AuditActor{UserID: &uid, Identity: "admin@example.com"} +} + +// TestSaveAccountServiceOverride_WritesThroughAuditedTx: the PUT writes via +// the audited store method inside a transaction, naming the session's actor, +// and never through the unaudited SaveAccountServiceOverride. +func TestSaveAccountServiceOverride_WritesThroughAuditedTx(t *testing.T) { + ctx := context.Background() + mockAuth := new(MockAuthService) + setupAdminAuth(ctx, mockAuth) + store := new(MockConfigStore) + var gotAudit config.AuditContext + store.On("SaveAccountServiceOverrideAuditedTx", ctx, mock.Anything, mock.AnythingOfType("*config.AccountServiceOverride"), mock.Anything). + Run(func(args mock.Arguments) { gotAudit = args.Get(3).(config.AuditContext) }).Return(nil) + h := &Handler{auth: mockAuth, config: store} + + _, err := h.saveAccountServiceOverride(ctx, adminRequest(`{"term":1}`), "11111111-1111-1111-1111-111111111111/service-overrides/aws/ec2") + require.NoError(t, err) + assert.Equal(t, config.AuditContext{Actor: adminAuditActor()}, gotAudit) + store.AssertCalled(t, "WithTx", ctx, mock.Anything) + store.AssertNotCalled(t, "SaveAccountServiceOverride", mock.Anything, mock.Anything) +} + +// TestSaveAccountServiceOverride_AuditFailureFailsTheSave: a failed audit +// write is a failed save, not a silent unaudited one. +func TestSaveAccountServiceOverride_AuditFailureFailsTheSave(t *testing.T) { + ctx := context.Background() + mockAuth := new(MockAuthService) + setupAdminAuth(ctx, mockAuth) + store := new(MockConfigStore) + store.On("SaveAccountServiceOverrideAuditedTx", ctx, mock.Anything, mock.Anything, mock.Anything). + Return(errors.New("failed to record config audit event")) + h := &Handler{auth: mockAuth, config: store} + + _, err := h.saveAccountServiceOverride(ctx, adminRequest(`{"term":1}`), "11111111-1111-1111-1111-111111111111/service-overrides/aws/ec2") + require.ErrorContains(t, err, "config audit event") +} + +func TestDeleteAccountServiceOverride_WritesThroughAuditedTx(t *testing.T) { + ctx := context.Background() + mockAuth := new(MockAuthService) + setupAdminAuth(ctx, mockAuth) + store := new(MockConfigStore) + store.On("DeleteAccountServiceOverrideAuditedTx", ctx, mock.Anything, + "11111111-1111-1111-1111-111111111111", "aws", "ec2", config.AuditContext{Actor: adminAuditActor()}).Return(nil) + h := &Handler{auth: mockAuth, config: store} + + _, err := h.deleteAccountServiceOverride(ctx, adminRequest(""), "11111111-1111-1111-1111-111111111111/service-overrides/aws/ec2") + require.NoError(t, err) + store.AssertExpectations(t) + store.AssertCalled(t, "WithTx", ctx, mock.Anything) + store.AssertNotCalled(t, "DeleteAccountServiceOverride", mock.Anything, mock.Anything, mock.Anything, mock.Anything) +} + +// TestImportOverrides_RecordsLinkedAuditEvents: an applied import records the +// import event (mode, counts) first, then one audited write per inserted or +// updated row, each linked to it and carrying its file row number, in key +// order rather than file order. +func TestImportOverrides_RecordsLinkedAuditEvents(t *testing.T) { + store := newOverrideTransferStore() + h := transferHandler(t, store, nil) + rows := []OverrideTransferRow{ + {AccountID: xferAWS, Provider: "aws", Service: "rds", Coverage: ptr(60.0)}, // update + {AccountID: xferAWS, Provider: "aws", Service: "elasticache", Term: ptr(1)}, // insert + {AccountID: xferAzure, Provider: "azure", Service: "vm", Enabled: ptr(false)}, // unchanged + {AccountID: xferAWS, Provider: "aws", Service: "opensearch", Term: ptr(7)}, // rejected + {AccountID: xferAWS, Provider: "aws", Service: "ec2", Enabled: ptr(true), Payment: ptr("bad")}, // rejected + } + + report, err := runImport(t, h, OverrideImportRequest{Mode: overrideImportBestEffort, Overrides: rows}) + require.NoError(t, err) + require.True(t, report.Applied) + + require.Len(t, store.events, 3) + imp := store.events[0].event + require.NotNil(t, imp, "the import event is written first") + assert.Equal(t, config.AuditOverrideImport, imp.EventType) + assert.Equal(t, adminAuditActor(), imp.Actor) + var details map[string]any + require.NoError(t, json.Unmarshal(imp.Details, &details)) + assert.Equal(t, "best_effort", details["mode"]) + assert.Equal(t, 5.0, details["rows"]) + assert.Equal(t, map[string]any{"insert": 1.0, "update": 1.0, "unchanged": 1.0, "rejected": 2.0}, details["summary"]) + + services := make([]string, 0, len(store.events)-1) + for _, e := range store.events[1:] { + require.NotNil(t, e.row) + require.NotNil(t, e.audit.ParentEventID) + assert.Equal(t, imp.ID, *e.audit.ParentEventID) + assert.Equal(t, adminAuditActor(), e.audit.Actor) + services = append(services, e.row.Service) + } + assert.Equal(t, []string{"elasticache", "rds"}, services, "rows are written in key order") + assert.Equal(t, map[string]any{"import_row": 2}, store.events[1].audit.Details) + assert.Equal(t, map[string]any{"import_row": 1}, store.events[2].audit.Details) +} + +// TestImportOverrides_NoAuditEventsWithoutWrites: a dry run, a rejected +// strict import and an import rolled back by a store failure leave no events. +func TestImportOverrides_NoAuditEventsWithoutWrites(t *testing.T) { + t.Run("dry run", func(t *testing.T) { + store := newOverrideTransferStore() + _, err := runImport(t, transferHandler(t, store, nil), OverrideImportRequest{DryRun: true, Overrides: fortyImportRows()}) + require.NoError(t, err) + assert.Empty(t, store.events) + }) + t.Run("strict with a rejected row", func(t *testing.T) { + store := newOverrideTransferStore() + rows := fortyImportRows() + rows[39].Term = ptr(7) + _, err := runImport(t, transferHandler(t, store, nil), OverrideImportRequest{Overrides: rows}) + require.Error(t, err) + assert.Empty(t, store.events) + }) + t.Run("store failure", func(t *testing.T) { + store := newOverrideTransferStore() + store.failSave = func(o *config.AccountServiceOverride) error { + if o.Service == "svc-40" { + return errors.New("connection reset") + } + return nil + } + _, err := runImport(t, transferHandler(t, store, nil), OverrideImportRequest{Overrides: fortyImportRows()}) + require.Error(t, err) + assert.Empty(t, store.events, "the rollback takes the import event and rows 1-39's events with it") + }) +} + +// TestImportOverrides_ImportEventFailureWritesNothing: no row is written when +// the import event cannot be recorded. +func TestImportOverrides_ImportEventFailureWritesNothing(t *testing.T) { + store := newOverrideTransferStore() + store.failEvent = errors.New("audit insert failed") + + _, err := runImport(t, transferHandler(t, store, nil), OverrideImportRequest{Overrides: fortyImportRows()}) + require.ErrorContains(t, err, "audit insert failed") + assert.Empty(t, store.saved) +} + +func TestConfigAuditActor(t *testing.T) { + user := configAuditActor(&Session{UserID: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", Email: "a@example.com", UserAPIKeyID: "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"}) + require.NotNil(t, user.UserID) + require.NotNil(t, user.APIKeyID) + assert.Equal(t, "a@example.com", user.Identity) + assert.Equal(t, "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb", *user.APIKeyID) + + admin := configAuditActor(&Session{UserID: apiKeyAdminUserID}) + assert.Nil(t, admin.UserID, "the admin API key's sentinel is not a user ID") + assert.Nil(t, admin.APIKeyID) + assert.Equal(t, apiKeyAdminUserID, admin.Identity) +} diff --git a/internal/api/handler_override_import.go b/internal/api/handler_override_import.go index 007e76196..c07ed2b5b 100644 --- a/internal/api/handler_override_import.go +++ b/internal/api/handler_override_import.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "math" + "sort" "strconv" "strings" "time" @@ -117,7 +118,7 @@ func (h *Handler) importAccountServiceOverrides(ctx context.Context, req *events if in.DryRun { return report, nil } - if err := h.applyOverrideImport(ctx, plan); err != nil { + if err := h.applyOverrideImport(ctx, plan, report, configAuditActor(session)); err != nil { return nil, err } report.Applied = true @@ -317,17 +318,35 @@ func (h *Handler) findStoredOverride(ctx context.Context, cache map[string][]con } // applyOverrideImport writes the insert and update rows in one transaction: -// either every planned row is written or, on any store failure, none is. -func (h *Handler) applyOverrideImport(ctx context.Context, plan []plannedOverride) error { +// either every planned row is written or, on any store failure, none is. The +// transaction also records the import's config_audit_events row (mode and +// counts) and, linked to it, one event per written row (issue #1493), so the +// audit trail commits or rolls back with the writes. +// +// Rows are written in (account, provider, service) order, not file order: +// each write locks its override's key until commit, and two imports that +// touch the same overrides must take those locks in the same order or they +// could deadlock. +func (h *Handler) applyOverrideImport(ctx context.Context, plan []plannedOverride, report *OverrideImportReport, actor config.AuditActor) error { now := time.Now() + ordered := append([]plannedOverride(nil), plan...) + sort.Slice(ordered, func(i, j int) bool { return plannedOverrideKey(ordered[i]) < plannedOverrideKey(ordered[j]) }) err := h.config.WithTx(ctx, func(tx pgx.Tx) error { - for _, p := range plan { + importEvent, err := overrideImportEvent(report, actor) + if err != nil { + return err + } + if err := h.config.SaveConfigAuditEventTx(ctx, tx, importEvent); err != nil { + return err + } + for _, p := range ordered { o := &config.AccountServiceOverride{ID: uuid.New().String(), AccountID: p.account.ID, Provider: p.row.Provider, Service: p.row.Service, CreatedAt: now, UpdatedAt: now} if p.existing != nil { o.ID, o.CreatedAt = p.existing.ID, p.existing.CreatedAt } applyOverrideFieldsTo(o, p.row.fields()) - if err := h.config.SaveAccountServiceOverrideTx(ctx, tx, o); err != nil { + audit := config.AuditContext{Actor: actor, ParentEventID: &importEvent.ID, Details: map[string]any{"import_row": p.rowNum}} + if err := h.config.SaveAccountServiceOverrideAuditedTx(ctx, tx, o, audit); err != nil { return fmt.Errorf("row %d: %w", p.rowNum, err) } } @@ -338,3 +357,38 @@ func (h *Handler) applyOverrideImport(ctx context.Context, plan []plannedOverrid } return nil } + +// plannedOverrideKey is the lock key order applyOverrideImport writes in. The +// fields are joined with NUL, which none of them can contain, so the order is +// the tuple order. +func plannedOverrideKey(p plannedOverride) string { + return p.account.ID + "\x00" + p.row.Provider + "\x00" + p.row.Service +} + +// overrideImportEvent is the import-level audit event: the mode and the +// per-action row counts of the applied import. +func overrideImportEvent(report *OverrideImportReport, actor config.AuditActor) (*config.AuditEvent, error) { + details, err := json.Marshal(map[string]any{"mode": report.Mode, "summary": report.Summary, "rows": len(report.Rows)}) + if err != nil { + return nil, fmt.Errorf("encode import audit details: %w", err) + } + return &config.AuditEvent{ + ID: uuid.New().String(), + EventType: config.AuditOverrideImport, + Actor: actor, + Details: details, + }, nil +} + +// configAuditActor names the session in a config audit event. The user and +// API key IDs are kept only when they are UUIDs (the stateless admin API key +// session carries a sentinel user ID); Identity always names the actor, +// falling back to that sentinel, as the 4-eyes comparison does. +func configAuditActor(session *Session) config.AuditActor { + actor := config.AuditActor{UserID: resolveCreatorUserID(session), Identity: fourEyesActorIdentity(session)} + if validateUUID(session.UserAPIKeyID) == nil { + keyID := session.UserAPIKeyID + actor.APIKeyID = &keyID + } + return actor +} diff --git a/internal/api/handler_override_transfer_test.go b/internal/api/handler_override_transfer_test.go index ddfb5d6c1..a8e5a8cff 100644 --- a/internal/api/handler_override_transfer_test.go +++ b/internal/api/handler_override_transfer_test.go @@ -22,13 +22,25 @@ import ( // overrideTransferStore is an in-memory account and override store for the // export/import tests (issue #9). Like Postgres, it refuses a ramp_schedule // longer than the VARCHAR(32) column, and WithTx undoes every write of a -// failed transaction. failSave, when set, injects a store failure. +// failed transaction, audit events included. failSave, when set, injects a +// store failure; failEvent fails SaveConfigAuditEventTx. events holds the +// import events and, for each audited row write, an event carrying the row's +// audit context (issue #1493). type overrideTransferStore struct { *MockConfigStore accounts []config.CloudAccount overrides map[string][]config.AccountServiceOverride saved []config.AccountServiceOverride + events []overrideTransferEvent failSave func(*config.AccountServiceOverride) error + failEvent error +} + +// overrideTransferEvent is one audit write the in-memory store received. +type overrideTransferEvent struct { + event *config.AuditEvent // set for SaveConfigAuditEventTx + row *config.AccountServiceOverride // set for an audited row write + audit config.AuditContext } func (s *overrideTransferStore) WithTx(_ context.Context, fn func(pgx.Tx) error) error { @@ -37,15 +49,33 @@ func (s *overrideTransferStore) WithTx(_ context.Context, fn func(pgx.Tx) error) overrides[k] = append([]config.AccountServiceOverride(nil), v...) } saved := append([]config.AccountServiceOverride(nil), s.saved...) + events := append([]overrideTransferEvent(nil), s.events...) if err := fn(nil); err != nil { - s.overrides, s.saved = overrides, saved + s.overrides, s.saved, s.events = overrides, saved, events + return err + } + return nil +} + +func (s *overrideTransferStore) SaveAccountServiceOverrideAuditedTx(ctx context.Context, _ pgx.Tx, o *config.AccountServiceOverride, audit config.AuditContext) error { + if err := s.SaveAccountServiceOverride(ctx, o); err != nil { return err } + row := *o + s.events = append(s.events, overrideTransferEvent{row: &row, audit: audit}) return nil } -func (s *overrideTransferStore) SaveAccountServiceOverrideTx(ctx context.Context, _ pgx.Tx, o *config.AccountServiceOverride) error { - return s.SaveAccountServiceOverride(ctx, o) +func (s *overrideTransferStore) SaveConfigAuditEventTx(_ context.Context, _ pgx.Tx, ev *config.AuditEvent) error { + if s.failEvent != nil { + return s.failEvent + } + if ev.ID == "" { + ev.ID = "import-event" + } + copied := *ev + s.events = append(s.events, overrideTransferEvent{event: &copied}) + return nil } func (s *overrideTransferStore) ListCloudAccounts(context.Context, config.CloudAccountFilter) ([]config.CloudAccount, error) { diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 4046e1969..74844db1b 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -606,10 +606,22 @@ type StoreInterface interface { // Account service overrides GetAccountServiceOverride(ctx context.Context, accountID, provider, service string) (*AccountServiceOverride, error) SaveAccountServiceOverride(ctx context.Context, override *AccountServiceOverride) error - // SaveAccountServiceOverrideTx is SaveAccountServiceOverride inside the - // caller's WithTx transaction. - SaveAccountServiceOverrideTx(ctx context.Context, tx pgx.Tx, override *AccountServiceOverride) error DeleteAccountServiceOverride(ctx context.Context, accountID, provider, service string) error + // SaveAccountServiceOverrideAuditedTx upserts the override inside the + // caller's WithTx transaction and records the change (the stored row + // before and after) in config_audit_events, issue #1493. It serializes on + // the override's key, so a caller writing several overrides in one tx must + // write them in a consistent key order. A write that changes no setting + // records nothing. override is updated to the stored row. + SaveAccountServiceOverrideAuditedTx(ctx context.Context, tx pgx.Tx, override *AccountServiceOverride, audit AuditContext) error + // DeleteAccountServiceOverrideAuditedTx deletes the override inside the + // caller's transaction and records the deleted row in config_audit_events. + // Deleting a missing override records nothing. + DeleteAccountServiceOverrideAuditedTx(ctx context.Context, tx pgx.Tx, accountID, provider, service string, audit AuditContext) error + // SaveConfigAuditEventTx records a config_audit_events row inside the + // caller's transaction (the override import's own event). An empty ID is + // filled in. + SaveConfigAuditEventTx(ctx context.Context, tx pgx.Tx, event *AuditEvent) error ListAccountServiceOverrides(ctx context.Context, accountID string) ([]AccountServiceOverride, error) // Plan ↔ account association diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index 2d6ade951..e37726f4d 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -4298,13 +4298,6 @@ func (s *PostgresStore) SaveAccountServiceOverride(ctx context.Context, o *Accou return saveAccountServiceOverride(ctx, s.db, o) } -// SaveAccountServiceOverrideTx upserts an account service override inside the -// caller's transaction, so a batch of writes (the override import) commits or -// rolls back as one. -func (s *PostgresStore) SaveAccountServiceOverrideTx(ctx context.Context, tx pgx.Tx, o *AccountServiceOverride) error { - return saveAccountServiceOverride(ctx, tx, o) -} - // overrideExecutor is satisfied by both the pool and a pgx.Tx. type overrideExecutor interface { Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) diff --git a/internal/config/store_postgres_config_audit.go b/internal/config/store_postgres_config_audit.go new file mode 100644 index 000000000..792159f60 --- /dev/null +++ b/internal/config/store_postgres_config_audit.go @@ -0,0 +1,279 @@ +package config + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" +) + +// Configuration audit trail (issue #1493, migration 000147). Every write to +// account_service_overrides records a config_audit_events row in the same +// transaction, so a change and its audit row commit or roll back together. + +// Config audit event types. +const ( + AuditOverrideUpsert = "account_service_override.upsert" + AuditOverrideDelete = "account_service_override.delete" + AuditOverrideImport = "account_service_override.import" +) + +// accountServiceOverrideLockPrefix namespaces the per-override advisory lock +// SaveAccountServiceOverrideAuditedTx takes. It is a distinct text input to +// the same hashtextextended(text, 0) bigint key space as the other +// hashtextextended locks; a 64-bit hash collision could only cause a spurious +// wait, never a correctness break. +const accountServiceOverrideLockPrefix = "cudly:account-service-override:" + +// AuditActor is who made a change. Identity is never empty: the email, +// or the admin API key actor. UserID and APIKeyID are set only when they are +// real users/API key IDs. +type AuditActor struct { + UserID *string + APIKeyID *string + Identity string +} + +// AuditContext is passed with each audited override write: the actor, +// the import event the write belongs to (nil for a single PUT or DELETE) and +// optional event details (an import row's number). +type AuditContext struct { + Actor AuditActor + ParentEventID *string + Details map[string]any +} + +// AuditEvent is one config_audit_events row. Before and After are the +// override settings as JSON (nil for none). +type AuditEvent struct { + ID string + EventType string + ParentEventID *string + Actor AuditActor + AccountID *string + Provider *string + Service *string + Before json.RawMessage + After json.RawMessage + Details json.RawMessage + CreatedAt time.Time +} + +// overrideAuditSnapshot is the before/after JSON of an override: its +// settings, without the identifiers and timestamps the event carries itself. +type overrideAuditSnapshot struct { + Enabled *bool `json:"enabled"` + Term *int `json:"term"` + Payment *string `json:"payment"` + Coverage *float64 `json:"coverage"` + RampSchedule *string `json:"ramp_schedule"` + IncludeEngines []string `json:"include_engines"` + ExcludeEngines []string `json:"exclude_engines"` + IncludeRegions []string `json:"include_regions"` + ExcludeRegions []string `json:"exclude_regions"` + IncludeTypes []string `json:"include_types"` + ExcludeTypes []string `json:"exclude_types"` +} + +// overrideSnapshotJSON returns o's settings as JSON, or nil for a nil o. +func overrideSnapshotJSON(o *AccountServiceOverride) (json.RawMessage, error) { + if o == nil { + return nil, nil + } + b, err := json.Marshal(overrideAuditSnapshot{ + Enabled: o.Enabled, Term: o.Term, Payment: o.Payment, Coverage: o.Coverage, RampSchedule: o.RampSchedule, + IncludeEngines: o.IncludeEngines, ExcludeEngines: o.ExcludeEngines, + IncludeRegions: o.IncludeRegions, ExcludeRegions: o.ExcludeRegions, + IncludeTypes: o.IncludeTypes, ExcludeTypes: o.ExcludeTypes, + }) + if err != nil { + return nil, fmt.Errorf("failed to encode service override audit snapshot: %w", err) + } + return b, nil +} + +const accountServiceOverrideColumns = `id, account_id, provider, service, + enabled, term, payment, coverage, ramp_schedule, + include_engines, exclude_engines, include_regions, exclude_regions, + include_types, exclude_types, + created_at, updated_at` + +// scanAccountServiceOverride scans one row of accountServiceOverrideColumns. +// A missing row is (nil, nil). +func scanAccountServiceOverride(row pgx.Row) (*AccountServiceOverride, error) { + var o AccountServiceOverride + err := row.Scan( + &o.ID, &o.AccountID, &o.Provider, &o.Service, + &o.Enabled, &o.Term, &o.Payment, &o.Coverage, &o.RampSchedule, + &o.IncludeEngines, &o.ExcludeEngines, &o.IncludeRegions, &o.ExcludeRegions, + &o.IncludeTypes, &o.ExcludeTypes, + &o.CreatedAt, &o.UpdatedAt, + ) + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, err + } + return &o, nil +} + +// SaveAccountServiceOverrideAuditedTx upserts o inside tx and records the +// change in config_audit_events. It takes a transaction-scoped advisory lock +// on the override's (account, provider, service), then reads the stored row +// FOR UPDATE, so the event's before is exactly the row this write replaces +// even when another writer creates or changes the same override at the same +// moment (a row lock alone cannot serialize two creates of a row that does +// not exist yet). after is read back from the upsert, so it is what the +// columns hold (coverage rounded to DECIMAL(5,2)). A write that changes no +// setting records no event. o is updated to the stored row (its ID is the +// existing override's when one was replaced). +// +// Callers writing several overrides in one tx must call this in a consistent +// key order, or two such transactions could deadlock on each other's locks. +func (s *PostgresStore) SaveAccountServiceOverrideAuditedTx(ctx context.Context, tx pgx.Tx, o *AccountServiceOverride, audit AuditContext) error { + key := accountServiceOverrideLockPrefix + o.AccountID + "/" + o.Provider + "/" + o.Service + if _, err := tx.Exec(ctx, "SELECT pg_advisory_xact_lock(hashtextextended($1, 0))", key); err != nil { + return fmt.Errorf("failed to lock service override: %w", err) + } + before, err := scanAccountServiceOverride(tx.QueryRow(ctx, + `SELECT `+accountServiceOverrideColumns+` FROM account_service_overrides + WHERE account_id = $1 AND provider = $2 AND service = $3 FOR UPDATE`, + o.AccountID, o.Provider, o.Service)) + if err != nil { + return fmt.Errorf("failed to read service override: %w", err) + } + after, err := upsertAccountServiceOverrideReturning(ctx, tx, o) + if err != nil { + return err + } + *o = *after + return recordOverrideChange(ctx, tx, AuditOverrideUpsert, o.AccountID, o.Provider, o.Service, before, after, audit) +} + +// upsertAccountServiceOverrideReturning is saveAccountServiceOverride with the +// stored row read back. +func upsertAccountServiceOverrideReturning(ctx context.Context, tx pgx.Tx, o *AccountServiceOverride) (*AccountServiceOverride, error) { + if o.ID == "" { + o.ID = uuid.New().String() + } + now := time.Now() + if o.CreatedAt.IsZero() { + o.CreatedAt = now + } + stored, err := scanAccountServiceOverride(tx.QueryRow(ctx, ` + INSERT INTO account_service_overrides (`+accountServiceOverrideColumns+`) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17) + ON CONFLICT (account_id, provider, service) DO UPDATE SET + enabled = $5, term = $6, payment = $7, coverage = $8, ramp_schedule = $9, + include_engines = $10, exclude_engines = $11, + include_regions = $12, exclude_regions = $13, + include_types = $14, exclude_types = $15, + updated_at = NOW() + RETURNING `+accountServiceOverrideColumns, + o.ID, o.AccountID, o.Provider, o.Service, + o.Enabled, o.Term, o.Payment, o.Coverage, o.RampSchedule, + o.IncludeEngines, o.ExcludeEngines, o.IncludeRegions, o.ExcludeRegions, + o.IncludeTypes, o.ExcludeTypes, + o.CreatedAt, now, + )) + if err != nil { + return nil, fmt.Errorf("failed to save service override: %w", err) + } + if stored == nil { + return nil, fmt.Errorf("failed to save service override: no row returned") + } + return stored, nil +} + +// DeleteAccountServiceOverrideAuditedTx deletes the override inside tx and +// records the deleted row as the event's before. Deleting an override that +// does not exist records nothing. +func (s *PostgresStore) DeleteAccountServiceOverrideAuditedTx(ctx context.Context, tx pgx.Tx, accountID, provider, service string, audit AuditContext) error { + before, err := scanAccountServiceOverride(tx.QueryRow(ctx, + `DELETE FROM account_service_overrides WHERE account_id = $1 AND provider = $2 AND service = $3 + RETURNING `+accountServiceOverrideColumns, + accountID, provider, service)) + if err != nil { + return fmt.Errorf("failed to delete service override: %w", err) + } + return recordOverrideChange(ctx, tx, AuditOverrideDelete, accountID, provider, service, before, nil, audit) +} + +// recordOverrideChange writes the event for one override write, unless the +// write changed no setting. +func recordOverrideChange(ctx context.Context, tx pgx.Tx, eventType, accountID, provider, service string, before, after *AccountServiceOverride, audit AuditContext) error { + beforeJSON, err := overrideSnapshotJSON(before) + if err != nil { + return err + } + afterJSON, err := overrideSnapshotJSON(after) + if err != nil { + return err + } + if bytes.Equal(beforeJSON, afterJSON) { + return nil // nothing changed (or nothing to delete) + } + details, err := auditDetailsJSON(audit.Details) + if err != nil { + return err + } + return insertConfigAuditEvent(ctx, tx, &AuditEvent{ + EventType: eventType, ParentEventID: audit.ParentEventID, Actor: audit.Actor, + AccountID: &accountID, Provider: &provider, Service: &service, + Before: beforeJSON, After: afterJSON, Details: details, + }) +} + +func auditDetailsJSON(details map[string]any) (json.RawMessage, error) { + if len(details) == 0 { + return nil, nil + } + b, err := json.Marshal(details) + if err != nil { + return nil, fmt.Errorf("failed to encode audit event details: %w", err) + } + return b, nil +} + +// SaveConfigAuditEventTx records ev inside tx. An empty ID is filled in, so a +// caller can link later events to it via ParentEventID. +func (s *PostgresStore) SaveConfigAuditEventTx(ctx context.Context, tx pgx.Tx, ev *AuditEvent) error { + return insertConfigAuditEvent(ctx, tx, ev) +} + +func insertConfigAuditEvent(ctx context.Context, tx pgx.Tx, ev *AuditEvent) error { + if ev.Actor.Identity == "" { + return fmt.Errorf("config audit event %s has no actor", ev.EventType) + } + if ev.ID == "" { + ev.ID = uuid.New().String() + } + err := tx.QueryRow(ctx, ` + INSERT INTO config_audit_events ( + id, event_type, parent_event_id, actor, actor_user_id, actor_api_key_id, + account_id, provider, service, before, after, details + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) + RETURNING created_at`, + ev.ID, ev.EventType, ev.ParentEventID, ev.Actor.Identity, ev.Actor.UserID, ev.Actor.APIKeyID, + ev.AccountID, ev.Provider, ev.Service, nullJSON(ev.Before), nullJSON(ev.After), nullJSON(ev.Details), + ).Scan(&ev.CreatedAt) + if err != nil { + return fmt.Errorf("failed to record config audit event: %w", err) + } + return nil +} + +// nullJSON passes an empty JSON value as SQL NULL rather than an empty +// string, which JSONB would refuse. +func nullJSON(raw json.RawMessage) any { + if len(raw) == 0 { + return nil + } + return string(raw) +} diff --git a/internal/config/store_postgres_config_audit_integration_test.go b/internal/config/store_postgres_config_audit_integration_test.go new file mode 100644 index 000000000..1d6b747f4 --- /dev/null +++ b/internal/config/store_postgres_config_audit_integration_test.go @@ -0,0 +1,135 @@ +//go:build integration + +package config + +import ( + "context" + "encoding/json" + "errors" + "testing" + + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" +) + +// Deterministic lock tests for SaveAccountServiceOverrideAuditedTx (issue +// #1493). Each parks a competing writer mid-transaction, waits until the +// audited save is blocked on a lock, commits the writer, and checks the +// event's before is the row the save actually replaced. + +type overrideAuditStoreFixture struct { + store *PostgresStore + account string +} + +func newOverrideAuditStoreFixture(ctx context.Context, t *testing.T) *overrideAuditStoreFixture { + t.Helper() + store := NewPostgresStore(testhelpers.RequireMigratedDatabase(ctx, t, getTestMigrationsPath()).DB) + acct := &CloudAccount{Name: "audit lock fixture", Provider: "aws", ExternalID: "123456789012", Enabled: true} + require.NoError(t, store.CreateCloudAccount(ctx, acct)) + return &overrideAuditStoreFixture{store: store, account: acct.ID} +} + +func (f *overrideAuditStoreFixture) hold(ctx context.Context, t *testing.T) pgx.Tx { + t.Helper() + tx, err := f.store.db.Begin(ctx) + require.NoError(t, err) + t.Cleanup(func() { + if rbErr := tx.Rollback(ctx); rbErr != nil && !errors.Is(rbErr, pgx.ErrTxClosed) { + t.Logf("writer rollback: %v", rbErr) + } + }) + return tx +} + +// saveInBackground runs an audited save of term=3 for aws/rds. +func (f *overrideAuditStoreFixture) saveInBackground(ctx context.Context) <-chan error { + done := make(chan error, 1) + go func() { + done <- f.store.WithTx(ctx, func(tx pgx.Tx) error { + o := &AccountServiceOverride{AccountID: f.account, Provider: "aws", Service: "rds", Term: intPtrAudit(3)} + return f.store.SaveAccountServiceOverrideAuditedTx(ctx, tx, o, AuditContext{Actor: AuditActor{Identity: "admin@example.com"}}) + }) + }() + return done +} + +// lastEventBefore returns the newest upsert event's before (nil for NULL). +func (f *overrideAuditStoreFixture) lastEventBefore(ctx context.Context, t *testing.T) map[string]any { + t.Helper() + var raw []byte + require.NoError(t, f.store.db.QueryRow(ctx, + `SELECT before FROM config_audit_events WHERE event_type = $1 ORDER BY seq DESC LIMIT 1`, AuditOverrideUpsert).Scan(&raw)) + if raw == nil { + return nil + } + var out map[string]any + require.NoError(t, json.Unmarshal(raw, &out)) + return out +} + +func intPtrAudit(v int) *int { return &v } + +// TestOverrideAudit_SaveQueuesBehindDelete: a delete removes the row while the +// save is starting. The save must wait for the row lock and then record a +// create (before NULL); reading the row before the delete committed would +// record the deleted row as replaced, so the trail would delete it twice. +func TestOverrideAudit_SaveQueuesBehindDelete(t *testing.T) { + ctx := context.Background() + f := newOverrideAuditStoreFixture(ctx, t) + require.NoError(t, f.store.SaveAccountServiceOverride(ctx, &AccountServiceOverride{AccountID: f.account, Provider: "aws", Service: "rds", Term: intPtrAudit(1)})) + + writer := f.hold(ctx, t) + _, err := writer.Exec(ctx, `DELETE FROM account_service_overrides WHERE account_id = $1`, f.account) + require.NoError(t, err) + done := f.saveInBackground(ctx) + testhelpers.WaitForLockWait(ctx, t, f.store.db, "%account_service_overrides%") + require.NoError(t, writer.Commit(ctx)) + require.NoError(t, testhelpers.Await(t, done)) + + assert.Nil(t, f.lastEventBefore(ctx, t), "the save created the override anew") +} + +// TestOverrideAudit_SaveQueuesBehindFirstCreate: another audited save holds +// the override's advisory lock and has inserted the row. A row lock cannot +// help (the row is not visible yet), so only the advisory lock makes this +// save wait and then record that row as its before. +func TestOverrideAudit_SaveQueuesBehindFirstCreate(t *testing.T) { + ctx := context.Background() + f := newOverrideAuditStoreFixture(ctx, t) + + writer := f.hold(ctx, t) + _, err := writer.Exec(ctx, "SELECT pg_advisory_xact_lock(hashtextextended($1, 0))", + accountServiceOverrideLockPrefix+f.account+"/aws/rds") + require.NoError(t, err) + require.NoError(t, saveAccountServiceOverride(ctx, writer, &AccountServiceOverride{AccountID: f.account, Provider: "aws", Service: "rds", Term: intPtrAudit(1)})) + done := f.saveInBackground(ctx) + testhelpers.WaitForLockWait(ctx, t, f.store.db, "") + require.NoError(t, writer.Commit(ctx)) + require.NoError(t, testhelpers.Await(t, done)) + + before := f.lastEventBefore(ctx, t) + require.NotNil(t, before, "the save replaced the first create's row") + assert.Equal(t, 1.0, before["term"]) +} + +// TestOverrideAudit_AfterIsTheStoredRow: after is read back from the write, +// so it shows what the DECIMAL(5,2) column holds, not the value sent. +func TestOverrideAudit_AfterIsTheStoredRow(t *testing.T) { + ctx := context.Background() + f := newOverrideAuditStoreFixture(ctx, t) + coverage := 33.333 + o := &AccountServiceOverride{AccountID: f.account, Provider: "aws", Service: "rds", Coverage: &coverage} + require.NoError(t, f.store.WithTx(ctx, func(tx pgx.Tx) error { + return f.store.SaveAccountServiceOverrideAuditedTx(ctx, tx, o, AuditContext{Actor: AuditActor{Identity: "admin@example.com"}}) + })) + var raw []byte + require.NoError(t, f.store.db.QueryRow(ctx, `SELECT after FROM config_audit_events`).Scan(&raw)) + var after map[string]any + require.NoError(t, json.Unmarshal(raw, &after)) + assert.Equal(t, 33.33, after["coverage"]) + assert.Equal(t, 33.33, *o.Coverage, "the caller's override is updated to the stored row") +} diff --git a/internal/config/store_postgres_config_audit_test.go b/internal/config/store_postgres_config_audit_test.go new file mode 100644 index 000000000..a80995ab4 --- /dev/null +++ b/internal/config/store_postgres_config_audit_test.go @@ -0,0 +1,152 @@ +package config + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/pashagolub/pgxmock/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +var overrideAuditCols = []string{ + "id", "account_id", "provider", "service", + "enabled", "term", "payment", "coverage", "ramp_schedule", + "include_engines", "exclude_engines", "include_regions", "exclude_regions", + "include_types", "exclude_types", "created_at", "updated_at", +} + +const auditTestAccount = "11111111-1111-1111-1111-111111111111" + +func overrideAuditRow(term int) *pgxmock.Rows { + at := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC) + return pgxmock.NewRows(overrideAuditCols).AddRow( + "o-1", auditTestAccount, "aws", "rds", + nil, &term, nil, nil, nil, + nil, nil, []string{"us-east-1"}, nil, nil, nil, at, at, + ) +} + +var testAuditCtx = AuditContext{Actor: AuditActor{Identity: "admin@example.com"}} + +// runAudited runs fn in a pgxmock transaction (BEGIN ... COMMIT expected by +// the caller). +func runAudited(t *testing.T, mock pgxmock.PgxPoolIface, fn func(tx pgx.Tx) error) error { + t.Helper() + return storeWith(mock).WithTx(context.Background(), fn) +} + +// expectOverrideLockedRead queues the advisory lock on the override's key and +// the FOR UPDATE read, in that order. +func expectOverrideLockedRead(mock pgxmock.PgxPoolIface, existing *pgxmock.Rows) { + mock.ExpectExec("pg_advisory_xact_lock").WithArgs(accountServiceOverrideLockPrefix + auditTestAccount + "/aws/rds"). + WillReturnResult(pgxmock.NewResult("SELECT", 1)) + mock.ExpectQuery(`FROM account_service_overrides\s+WHERE account_id = \$1 AND provider = \$2 AND service = \$3 FOR UPDATE`). + WithArgs(auditTestAccount, "aws", "rds").WillReturnRows(existing) +} + +// TestPGXMock_SaveOverrideAudited_RecordsBeforeAndAfter: lock, locked read, +// upsert read back, then the event with both snapshots, in one transaction. +func TestPGXMock_SaveOverrideAudited_RecordsBeforeAndAfter(t *testing.T) { + mock := newMock(t) + mock.ExpectBegin() + expectOverrideLockedRead(mock, overrideAuditRow(1)) + mock.ExpectQuery(`INSERT INTO account_service_overrides .* RETURNING`).WithArgs(anyArgsCfg(17)...). + WillReturnRows(overrideAuditRow(3)) + mock.ExpectQuery("INSERT INTO config_audit_events").WithArgs(anyArgsCfg(12)...). + WillReturnRows(pgxmock.NewRows([]string{"created_at"}).AddRow(time.Now())) + mock.ExpectCommit() + + o := &AccountServiceOverride{ID: "fresh-id", AccountID: auditTestAccount, Provider: "aws", Service: "rds", Term: intPtrUnit(3)} + require.NoError(t, runAudited(t, mock, func(tx pgx.Tx) error { + return (&PostgresStore{db: mock}).SaveAccountServiceOverrideAuditedTx(context.Background(), tx, o, testAuditCtx) + })) + assert.Equal(t, "o-1", o.ID, "the caller sees the stored row's ID") + assert.NoError(t, mock.ExpectationsWereMet()) +} + +// TestPGXMock_SaveOverrideAudited_NoChangeNoEvent: a save that leaves every +// setting as it was records no event. +func TestPGXMock_SaveOverrideAudited_NoChangeNoEvent(t *testing.T) { + mock := newMock(t) + mock.ExpectBegin() + expectOverrideLockedRead(mock, overrideAuditRow(1)) + mock.ExpectQuery(`INSERT INTO account_service_overrides .* RETURNING`).WithArgs(anyArgsCfg(17)...). + WillReturnRows(overrideAuditRow(1)) + // No INSERT INTO config_audit_events. + mock.ExpectCommit() + + o := &AccountServiceOverride{AccountID: auditTestAccount, Provider: "aws", Service: "rds", Term: intPtrUnit(1)} + require.NoError(t, runAudited(t, mock, func(tx pgx.Tx) error { + return (&PostgresStore{db: mock}).SaveAccountServiceOverrideAuditedTx(context.Background(), tx, o, testAuditCtx) + })) + assert.NoError(t, mock.ExpectationsWereMet()) +} + +// TestPGXMock_DeleteOverrideAudited: the deleted row is the event's before; a +// delete that finds nothing records nothing. +func TestPGXMock_DeleteOverrideAudited(t *testing.T) { + t.Run("row deleted", func(t *testing.T) { + mock := newMock(t) + mock.ExpectBegin() + mock.ExpectQuery(`DELETE FROM account_service_overrides .* RETURNING`).WithArgs(auditTestAccount, "aws", "rds"). + WillReturnRows(overrideAuditRow(1)) + mock.ExpectQuery("INSERT INTO config_audit_events"). + WithArgs(pgxmock.AnyArg(), AuditOverrideDelete, pgxmock.AnyArg(), "admin@example.com", pgxmock.AnyArg(), pgxmock.AnyArg(), + pgxmock.AnyArg(), pgxmock.AnyArg(), pgxmock.AnyArg(), pgxmock.AnyArg(), nil, nil). + WillReturnRows(pgxmock.NewRows([]string{"created_at"}).AddRow(time.Now())) + mock.ExpectCommit() + require.NoError(t, runAudited(t, mock, func(tx pgx.Tx) error { + return (&PostgresStore{db: mock}).DeleteAccountServiceOverrideAuditedTx(context.Background(), tx, auditTestAccount, "aws", "rds", testAuditCtx) + })) + assert.NoError(t, mock.ExpectationsWereMet()) + }) + t.Run("nothing to delete", func(t *testing.T) { + mock := newMock(t) + mock.ExpectBegin() + mock.ExpectQuery(`DELETE FROM account_service_overrides .* RETURNING`).WithArgs(auditTestAccount, "aws", "rds"). + WillReturnRows(pgxmock.NewRows(overrideAuditCols)) + mock.ExpectCommit() + require.NoError(t, runAudited(t, mock, func(tx pgx.Tx) error { + return (&PostgresStore{db: mock}).DeleteAccountServiceOverrideAuditedTx(context.Background(), tx, auditTestAccount, "aws", "rds", testAuditCtx) + })) + assert.NoError(t, mock.ExpectationsWereMet()) + }) +} + +// TestPGXMock_SaveConfigAuditEvent_RequiresActor: an event with no actor is +// refused before it reaches the database, and the transaction rolls back. +func TestPGXMock_SaveConfigAuditEvent_RequiresActor(t *testing.T) { + mock := newMock(t) + mock.ExpectBegin() + mock.ExpectRollback() + err := runAudited(t, mock, func(tx pgx.Tx) error { + return (&PostgresStore{db: mock}).SaveConfigAuditEventTx(context.Background(), tx, &AuditEvent{EventType: AuditOverrideImport}) + }) + require.ErrorContains(t, err, "no actor") + assert.NoError(t, mock.ExpectationsWereMet()) +} + +func TestOverrideSnapshotJSON(t *testing.T) { + got, err := overrideSnapshotJSON(&AccountServiceOverride{ + ID: "o-1", AccountID: auditTestAccount, Provider: "aws", Service: "rds", + Term: intPtrUnit(3), IncludeRegions: []string{"us-east-1"}, CreatedAt: time.Now(), + }) + require.NoError(t, err) + var m map[string]any + require.NoError(t, json.Unmarshal(got, &m)) + assert.Equal(t, 3.0, m["term"]) + assert.Equal(t, []any{"us-east-1"}, m["include_regions"]) + assert.Contains(t, m, "enabled", "an unset setting is recorded as null, not omitted") + assert.NotContains(t, m, "id") + assert.NotContains(t, m, "created_at") + + none, err := overrideSnapshotJSON(nil) + require.NoError(t, err) + assert.Nil(t, none) +} + +func intPtrUnit(v int) *int { return &v } diff --git a/internal/database/postgres/migrations/000147_config_audit_events.down.sql b/internal/database/postgres/migrations/000147_config_audit_events.down.sql new file mode 100644 index 000000000..439bd94c6 --- /dev/null +++ b/internal/database/postgres/migrations/000147_config_audit_events.down.sql @@ -0,0 +1,3 @@ +-- Reverse 000147: drop the configuration audit trail (issue #1493). +-- Lossy by design: every recorded event is dropped with the table. +DROP TABLE IF EXISTS config_audit_events; diff --git a/internal/database/postgres/migrations/000147_config_audit_events.up.sql b/internal/database/postgres/migrations/000147_config_audit_events.up.sql new file mode 100644 index 000000000..1e92e4953 --- /dev/null +++ b/internal/database/postgres/migrations/000147_config_audit_events.up.sql @@ -0,0 +1,70 @@ +-- Configuration audit trail (issue #1493). +-- +-- Writes to account_service_overrides (the single PUT and DELETE endpoints and +-- the bulk import) recorded nothing, so nobody could answer "who changed this +-- account's override, when, and from what". Each change now writes one row +-- here, in the same transaction as the change itself: +-- +-- account_service_override.upsert a PUT or an imported row created or +-- changed an override (before is NULL for +-- a create) +-- account_service_override.delete a DELETE removed one (after is NULL) +-- account_service_override.import an applied import; its mode and row +-- counts are in details, and every row it +-- wrote links back via parent_event_id +-- +-- A save that changes nothing, a delete of a missing override and a dry-run +-- import write no rows. +-- +-- before / after are the stored override's settings (enabled, term, payment, +-- coverage, ramp_schedule and the include/exclude filters) as JSON. +-- +-- No foreign keys to users, API keys or cloud accounts: the trail must outlive +-- what it names (an account delete cascades its overrides away). actor keeps +-- the human-readable identity (email, or the admin API key actor) for when the +-- referenced user row is gone. seq orders events: every event of one +-- transaction shares created_at. +CREATE TABLE IF NOT EXISTS config_audit_events ( + id UUID PRIMARY KEY, + seq BIGINT GENERATED ALWAYS AS IDENTITY UNIQUE, + event_type TEXT NOT NULL CHECK (event_type IN ( + 'account_service_override.upsert', + 'account_service_override.delete', + 'account_service_override.import' + )), + parent_event_id UUID REFERENCES config_audit_events (id), + actor TEXT NOT NULL CHECK (actor <> ''), + actor_user_id UUID, + actor_api_key_id UUID, + account_id UUID, + provider TEXT, + service TEXT, + before JSONB, + after JSONB, + details JSONB, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + -- A per-override event names its override and records a change. + CONSTRAINT config_audit_events_override_target_check CHECK ( + event_type = 'account_service_override.import' + OR (account_id IS NOT NULL AND provider IS NOT NULL AND service IS NOT NULL) + ), + CONSTRAINT config_audit_events_override_change_check CHECK ( + (event_type <> 'account_service_override.upsert' OR after IS NOT NULL) + AND (event_type <> 'account_service_override.delete' OR (before IS NOT NULL AND after IS NULL)) + ), + -- An import is the top of its tree. + CONSTRAINT config_audit_events_import_parent_check CHECK ( + event_type <> 'account_service_override.import' OR parent_event_id IS NULL + ) +); + +CREATE INDEX IF NOT EXISTS idx_config_audit_events_account + ON config_audit_events (account_id, seq DESC) + WHERE account_id IS NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_config_audit_events_parent + ON config_audit_events (parent_event_id) + WHERE parent_event_id IS NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_config_audit_events_created + ON config_audit_events (created_at DESC); diff --git a/internal/database/postgres/migrations/000147_config_audit_events_test.go b/internal/database/postgres/migrations/000147_config_audit_events_test.go new file mode 100644 index 000000000..92b21575d --- /dev/null +++ b/internal/database/postgres/migrations/000147_config_audit_events_test.go @@ -0,0 +1,102 @@ +//go:build integration + +package migrations_test + +import ( + "context" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" +) + +const ( + m147Account = "14700000-0000-4000-8000-000000000001" + m147Import = "14700000-0000-4000-8000-0000000000a1" +) + +func m147Insert(ctx context.Context, pool *pgxpool.Pool, id, eventType string, parent, account, before, after any, actor string) error { + _, err := pool.Exec(ctx, ` + INSERT INTO config_audit_events (id, event_type, parent_event_id, actor, account_id, provider, service, before, after) + VALUES ($1, $2, $3, $4, $5, CASE WHEN $5::uuid IS NULL THEN NULL ELSE 'aws' END, + CASE WHEN $5::uuid IS NULL THEN NULL ELSE 'rds' END, $6, $7)`, + id, eventType, parent, actor, account, before, after) + return err +} + +func m147TableExists(ctx context.Context, t *testing.T, pool *pgxpool.Pool) bool { + t.Helper() + var exists bool + require.NoError(t, pool.QueryRow(ctx, `SELECT to_regclass('config_audit_events') IS NOT NULL`).Scan(&exists)) + return exists +} + +// TestMigration_ConfigAuditEvents covers 000147 (issue #1493): the table +// accepts the three override event shapes, refuses malformed ones, and the +// down migration drops it. +func TestMigration_ConfigAuditEvents(t *testing.T) { + ctx := context.Background() + container := testhelpers.RequirePostgresContainer(ctx, t) + t.Cleanup(func() { require.NoError(t, container.Cleanup(context.Background())) }) + pool, path := container.DB.Pool(), getMigrationsPath() + + const version = 147 + prev := previousMigrationVersion(t, version) + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, prev)) + require.False(t, m147TableExists(ctx, t, pool)) + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, version)) + + const row = `{"term": 1}` + require.NoError(t, m147Insert(ctx, pool, m147Import, "account_service_override.import", nil, nil, nil, nil, "admin@example.com")) + require.NoError(t, m147Insert(ctx, pool, "14700000-0000-4000-8000-0000000000b1", "account_service_override.upsert", m147Import, m147Account, nil, row, "admin@example.com")) + require.NoError(t, m147Insert(ctx, pool, "14700000-0000-4000-8000-0000000000b2", "account_service_override.delete", nil, m147Account, row, nil, "admin@example.com")) + + rejected := map[string]func() error{ + "unknown event type": func() error { + return m147Insert(ctx, pool, "14700000-0000-4000-8000-0000000000c1", "global_config.update", nil, m147Account, nil, row, "admin@example.com") + }, + "empty actor": func() error { + return m147Insert(ctx, pool, "14700000-0000-4000-8000-0000000000c2", "account_service_override.upsert", nil, m147Account, nil, row, "") + }, + "override event without its override": func() error { + return m147Insert(ctx, pool, "14700000-0000-4000-8000-0000000000c3", "account_service_override.upsert", nil, nil, nil, row, "admin@example.com") + }, + "upsert without after": func() error { + return m147Insert(ctx, pool, "14700000-0000-4000-8000-0000000000c4", "account_service_override.upsert", nil, m147Account, row, nil, "admin@example.com") + }, + "delete with an after": func() error { + return m147Insert(ctx, pool, "14700000-0000-4000-8000-0000000000c5", "account_service_override.delete", nil, m147Account, row, row, "admin@example.com") + }, + "import under a parent": func() error { + return m147Insert(ctx, pool, "14700000-0000-4000-8000-0000000000c6", "account_service_override.import", m147Import, nil, nil, nil, "admin@example.com") + }, + "parent that does not exist": func() error { + return m147Insert(ctx, pool, "14700000-0000-4000-8000-0000000000c7", "account_service_override.upsert", "14700000-0000-4000-8000-0000000000ff", m147Account, nil, row, "admin@example.com") + }, + } + for name, insert := range rejected { + assert.Error(t, insert(), name) + } + + var seqs []int64 + rows, err := pool.Query(ctx, `SELECT seq FROM config_audit_events ORDER BY seq`) + require.NoError(t, err) + for rows.Next() { + var s int64 + require.NoError(t, rows.Scan(&s)) + seqs = append(seqs, s) + } + require.NoError(t, rows.Err()) + require.Len(t, seqs, 3) + assert.Less(t, seqs[0], seqs[1]) + assert.Less(t, seqs[1], seqs[2]) + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, prev)) + assert.False(t, m147TableExists(ctx, t, pool), "down drops the table") + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, version)) + assert.True(t, m147TableExists(ctx, t, pool), "re-up recreates it") +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index 9f42ece62..304cd3f95 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -1598,14 +1598,37 @@ func (m *MockConfigStore) SaveAccountServiceOverride(ctx context.Context, overri return args.Error(0) } -// SaveAccountServiceOverrideTx falls back to SaveAccountServiceOverride when -// no expectation is registered; see UpdatePurchasePlanTx. -func (m *MockConfigStore) SaveAccountServiceOverrideTx(ctx context.Context, tx pgx.Tx, override *config.AccountServiceOverride) error { - m.record("SaveAccountServiceOverrideTx", ctx, tx, override) - if !isExpected(&m.Mock, "SaveAccountServiceOverrideTx") { +// SaveAccountServiceOverrideAuditedTx falls back to SaveAccountServiceOverride +// when no expectation is registered (see UpdatePurchasePlanTx), so tests that +// seed SaveAccountServiceOverrideFn keep observing the write. +func (m *MockConfigStore) SaveAccountServiceOverrideAuditedTx(ctx context.Context, tx pgx.Tx, override *config.AccountServiceOverride, audit config.AuditContext) error { + m.record("SaveAccountServiceOverrideAuditedTx", ctx, tx, override, audit) + if !isExpected(&m.Mock, "SaveAccountServiceOverrideAuditedTx") { return m.SaveAccountServiceOverride(ctx, override) } - args := m.Called(ctx, tx, override) + args := m.Called(ctx, tx, override, audit) + return args.Error(0) +} + +// DeleteAccountServiceOverrideAuditedTx falls back to +// DeleteAccountServiceOverride when no expectation is registered. +func (m *MockConfigStore) DeleteAccountServiceOverrideAuditedTx(ctx context.Context, tx pgx.Tx, accountID, provider, service string, audit config.AuditContext) error { + m.record("DeleteAccountServiceOverrideAuditedTx", ctx, tx, accountID, provider, service, audit) + if !isExpected(&m.Mock, "DeleteAccountServiceOverrideAuditedTx") { + return m.DeleteAccountServiceOverride(ctx, accountID, provider, service) + } + args := m.Called(ctx, tx, accountID, provider, service, audit) + return args.Error(0) +} + +// SaveConfigAuditEventTx records the event and returns nil when no +// expectation is registered. +func (m *MockConfigStore) SaveConfigAuditEventTx(ctx context.Context, tx pgx.Tx, event *config.AuditEvent) error { + m.record("SaveConfigAuditEventTx", ctx, tx, event) + if !isExpected(&m.Mock, "SaveConfigAuditEventTx") { + return nil + } + args := m.Called(ctx, tx, event) return args.Error(0) }