From 5d2c3e749f3fcf29a2b7b80bcaa6647d36294c65 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Fri, 23 Oct 2026 13:34:53 +1000 Subject: [PATCH] feat(ec2): seal account settings with revision-guarded writes (#1241) Replace the unwired delete-based CleanupAccount with an owner-local sealed state in the existing account-settings KV value. All settings mutations now use revision-guarded create/update; a sealed record rejects every read and mutation with Blocked and cannot be recreated. Records without a state field remain active. SealAccount CAS-writes the sealed value whether the key is present, absent or tombstoned, and succeeds only on a leader-accepted write, returning its revision. It first checks, through a read-only IAM account-status reader, that the account is already TERMINATING; otherwise it refuses with IncorrectState and writes nothing. The daemon serves the internal subject ec2.v1.SealAccountSettings. Nothing calls it yet. --- contracts/ec2/v1/README.md | 22 + contracts/ec2/v1/account_settings_seal.go | 28 + .../ec2/v1/account_settings_seal_test.go | 43 ++ spinifex/daemon/daemon.go | 7 + .../daemon_handlers_account_settings_seal.go | 23 + ...mon_handlers_account_settings_seal_test.go | 101 ++++ spinifex/domains/ec2/account/cleanup.go | 90 --- spinifex/domains/ec2/account/cleanup_test.go | 181 ------ spinifex/domains/ec2/account/seal.go | 135 +++++ spinifex/domains/ec2/account/seal_test.go | 559 ++++++++++++++++++ spinifex/domains/ec2/account/service_impl.go | 148 +++-- .../domains/ec2/account/service_impl_test.go | 43 ++ .../handlers/iam/account_status_reader.go | 56 ++ .../iam/account_status_reader_test.go | 102 ++++ 14 files changed, 1212 insertions(+), 326 deletions(-) create mode 100644 contracts/ec2/v1/account_settings_seal.go create mode 100644 contracts/ec2/v1/account_settings_seal_test.go create mode 100644 spinifex/daemon/daemon_handlers_account_settings_seal.go create mode 100644 spinifex/daemon/daemon_handlers_account_settings_seal_test.go delete mode 100644 spinifex/domains/ec2/account/cleanup.go delete mode 100644 spinifex/domains/ec2/account/cleanup_test.go create mode 100644 spinifex/domains/ec2/account/seal.go create mode 100644 spinifex/domains/ec2/account/seal_test.go create mode 100644 spinifex/handlers/iam/account_status_reader.go create mode 100644 spinifex/handlers/iam/account_status_reader_test.go diff --git a/contracts/ec2/v1/README.md b/contracts/ec2/v1/README.md index cc3c9cf9f..4bf325e37 100644 --- a/contracts/ec2/v1/README.md +++ b/contracts/ec2/v1/README.md @@ -60,3 +60,25 @@ The subject name carries the contract major; `platform_inventory_test.go` preser Producer and consumer are in the coordinated Spinifex release, so the window is N and N-1. A daemon that predates the subject has no subscription, so the request gets no responder; the consumer treats no responder, a timeout, an error reply, an undecodable reply, a reply for another account, and a missing, repeated or unknown-status kind as `scan-incomplete`, never clean. An incompatible change to the subject, fields or vocabulary needs a new major. + +## Seal account settings + +`SealAccountSettingsSubject` (`ec2.v1.SealAccountSettings`) is a request/reply route on which a daemon in the `spinifex-workers` queue group irreversibly seals one account's EC2 settings record. +EC2 owns it: `domains/ec2/account` serves it through the daemon. +`accountteardown` is the intended consumer from a later release; no teardown calls it yet. + +The request names the account in its body (`account_id`) and may carry an opaque `correlation_ref` that is echoed for logs. +The daemon ignores `X-Account-ID` for this subject: the account sealed is the one named in the body. +A seal is refused with `IncorrectState` unless IAM already holds the account as `TERMINATING`; a precondition read failure answers `InternalError`. +Both refusals write nothing. + +A reply is sent only on success: `disposition` is always `sealed` and `revision` is the stream leader's accepted write for that seal. +A failure is the standard awserrors error payload, not this reply shape. +Sealing is idempotent: a seal on an already-sealed record re-writes the same value and returns a fresh, higher revision. + +### Compatibility boundary + +The subject name carries the contract major; `account_settings_seal_test.go` preserves the version-1 JSON shape. +Producer and consumer are in the coordinated Spinifex release, so the window is N and N-1. +A daemon that predates the subject has no subscription, so the request gets no responder. +An incompatible change to the subject or fields needs a new major. diff --git a/contracts/ec2/v1/account_settings_seal.go b/contracts/ec2/v1/account_settings_seal.go new file mode 100644 index 000000000..cd5ef3a47 --- /dev/null +++ b/contracts/ec2/v1/account_settings_seal.go @@ -0,0 +1,28 @@ +package ec2v1 + +// SealAccountSettingsSubject is the request/reply subject on which a daemon +// seals one account's EC2 settings record. Sealing is irreversible and refused +// unless IAM already holds the account as TERMINATING. +const SealAccountSettingsSubject = "ec2.v1.SealAccountSettings" + +// AccountSettingsDispositionSealed is the only disposition a successful reply +// carries: a reply is sent only on success, so there is no other value. +const AccountSettingsDispositionSealed = "sealed" + +// SealAccountSettingsRequest asks for one account's EC2 settings to be +// sealed. CorrelationRef is an opaque caller reference echoed for logs; it +// grants nothing and is never a fence or authority to write. +type SealAccountSettingsRequest struct { + AccountID string `json:"account_id"` + CorrelationRef string `json:"correlation_ref,omitempty"` +} + +// SealAccountSettingsReply confirms accountID's settings record was sealed at +// Revision, the stream leader's accepted write for that seal. A failure is +// the standard awserrors error payload, not this type. +type SealAccountSettingsReply struct { + AccountID string `json:"account_id"` + CorrelationRef string `json:"correlation_ref,omitempty"` + Disposition string `json:"disposition"` + Revision uint64 `json:"revision"` +} diff --git a/contracts/ec2/v1/account_settings_seal_test.go b/contracts/ec2/v1/account_settings_seal_test.go new file mode 100644 index 000000000..765450307 --- /dev/null +++ b/contracts/ec2/v1/account_settings_seal_test.go @@ -0,0 +1,43 @@ +package ec2v1 + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestSealAccountSettingsSubject(t *testing.T) { + require.Equal(t, "ec2.v1.SealAccountSettings", SealAccountSettingsSubject) +} + +func TestSealAccountSettingsRequestJSONContract(t *testing.T) { + data, err := json.Marshal(SealAccountSettingsRequest{AccountID: "111122223333", CorrelationRef: "req-1"}) + require.NoError(t, err) + require.JSONEq(t, `{"account_id":"111122223333","correlation_ref":"req-1"}`, string(data)) + + data, err = json.Marshal(SealAccountSettingsRequest{AccountID: "111122223333"}) + require.NoError(t, err) + require.JSONEq(t, `{"account_id":"111122223333"}`, string(data)) +} + +func TestSealAccountSettingsReplyJSONContract(t *testing.T) { + reply := SealAccountSettingsReply{ + AccountID: "111122223333", + CorrelationRef: "req-1", + Disposition: AccountSettingsDispositionSealed, + Revision: 42, + } + data, err := json.Marshal(reply) + require.NoError(t, err) + require.JSONEq(t, `{ + "account_id": "111122223333", + "correlation_ref": "req-1", + "disposition": "sealed", + "revision": 42 + }`, string(data)) + + data, err = json.Marshal(SealAccountSettingsReply{AccountID: "111122223333", Disposition: AccountSettingsDispositionSealed}) + require.NoError(t, err) + require.JSONEq(t, `{"account_id":"111122223333","disposition":"sealed","revision":0}`, string(data)) +} diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 001f041fb..65fae4395 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -1026,6 +1026,7 @@ func (d *Daemon) subscribeAll() error { {"ec2.GetSerialConsoleAccessStatus", handleNATSRequest(d.node, d.accountService.GetSerialConsoleAccessStatus), "spinifex-workers"}, {"ec2.EnableSerialConsoleAccess", handleNATSRequest(d.node, d.accountService.EnableSerialConsoleAccess), "spinifex-workers"}, {"ec2.DisableSerialConsoleAccess", handleNATSRequest(d.node, d.accountService.DisableSerialConsoleAccess), "spinifex-workers"}, + {ec2v1.SealAccountSettingsSubject, handleNATSRequest(d.node, d.sealAccountSettings), "spinifex-workers"}, {clusterv1.NodeHealthSubject(d.node), d.handleHealthCheck, ""}, {clusterv1.NodesDiscoverSubject, d.handleNodeDiscover, ""}, {clusterv1.NodeStatusSubject, d.handleNodeStatus, ""}, @@ -1992,6 +1993,12 @@ func (d *Daemon) startCluster() error { if err != nil { return fmt.Errorf("failed to initialize account settings service: %w", err) } + accountStatusJS, err := jetstream.New(d.natsConn) + if err != nil { + return fmt.Errorf("jetstream handle for account status reader: %w", err) + } + var _ ec2account.AccountTerminationChecker = (*handlers_iam.AccountStatusReader)(nil) + d.accountService.SetAccountTerminationChecker(handlers_iam.NewAccountStatusReader(accountStatusJS)) // ELBv2 resolves listener certificate private keys through its own ACM // Store over the same JetStream bucket the ACM service writes, so it needs diff --git a/spinifex/daemon/daemon_handlers_account_settings_seal.go b/spinifex/daemon/daemon_handlers_account_settings_seal.go new file mode 100644 index 000000000..f20a415a6 --- /dev/null +++ b/spinifex/daemon/daemon_handlers_account_settings_seal.go @@ -0,0 +1,23 @@ +package daemon + +import ( + "context" + + ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1" +) + +// sealAccountSettings serves the irreversible EC2 settings seal. The account +// comes from the request body, not X-Account-ID: the sealed account is part +// of the versioned contract, so the header is ignored. +func (d *Daemon) sealAccountSettings(ctx context.Context, req *ec2v1.SealAccountSettingsRequest, _ string) (*ec2v1.SealAccountSettingsReply, error) { + seal, err := d.accountService.SealAccount(ctx, req.AccountID) + if err != nil { + return nil, err + } + return &ec2v1.SealAccountSettingsReply{ + AccountID: req.AccountID, + CorrelationRef: req.CorrelationRef, + Disposition: ec2v1.AccountSettingsDispositionSealed, + Revision: seal.Revision, + }, nil +} diff --git a/spinifex/daemon/daemon_handlers_account_settings_seal_test.go b/spinifex/daemon/daemon_handlers_account_settings_seal_test.go new file mode 100644 index 000000000..b6f85a75d --- /dev/null +++ b/spinifex/daemon/daemon_handlers_account_settings_seal_test.go @@ -0,0 +1,101 @@ +package daemon + +import ( + "context" + "encoding/json" + "testing" + "time" + + ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1" + "github.com/mulgadc/spinifex/internal/testkit" + ec2account "github.com/mulgadc/spinifex/spinifex/domains/ec2/account" + awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// sealTestChecker answers AccountTerminating from a fixed set, standing in +// for IAM in these daemon-level tests. +type sealTestChecker struct { + terminating map[string]bool +} + +func (c *sealTestChecker) AccountTerminating(_ context.Context, accountID string) (bool, error) { + return c.terminating[accountID], nil +} + +// newSealTestDaemon builds a daemon serving its real subscription table over +// a real account settings service, wired with checker as its termination +// precondition. +func newSealTestDaemon(t *testing.T, checker ec2account.AccountTerminationChecker) (*Daemon, *nats.Conn) { + t.Helper() + ns, nc, _ := testutil.StartTestJetStream(t) + + acctSvc, err := ec2account.NewAccountSettingsServiceImplWithNATS(t.Context(), nil, nc) + require.NoError(t, err) + acctSvc.SetAccountTerminationChecker(checker) + + d := newSubscribeTestDaemon(t, nc, "", "") + d.accountService = acctSvc + require.NoError(t, d.subscribeAll()) + require.NoError(t, nc.Flush()) + + caller, err := nats.Connect(ns.ClientURL()) + require.NoError(t, err) + t.Cleanup(caller.Close) + return d, caller +} + +const ( + sealTestTerminating = "111122223333" + sealTestActive = "444455556666" +) + +func sealRequest(t *testing.T, nc *nats.Conn, body any, header string) []byte { + t.Helper() + data, err := json.Marshal(body) + require.NoError(t, err) + reply, err := natsRequestAs(nc, ec2v1.SealAccountSettingsSubject, data, header, 5*time.Second) + require.NoError(t, err) + return reply.Data +} + +// The header is ignored: the body account is what gets sealed, even when +// X-Account-ID names a different one. +func TestSealAccountSettingsSubjectSealsTheBodyAccount(t *testing.T) { + _, nc := newSealTestDaemon(t, &sealTestChecker{terminating: map[string]bool{sealTestTerminating: true}}) + + raw := sealRequest(t, nc, ec2v1.SealAccountSettingsRequest{AccountID: sealTestTerminating, CorrelationRef: "req-1"}, sealTestActive) + require.Empty(t, replyErrCode(t, raw), string(raw)) + + var reply ec2v1.SealAccountSettingsReply + require.NoError(t, json.Unmarshal(raw, &reply)) + assert.Equal(t, sealTestTerminating, reply.AccountID) + assert.Equal(t, "req-1", reply.CorrelationRef) + assert.Equal(t, ec2v1.AccountSettingsDispositionSealed, reply.Disposition) + assert.NotZero(t, reply.Revision) +} + +// Only an account IAM holds as TERMINATING may be sealed; ACTIVE is refused +// before any write. +func TestSealAccountSettingsSubjectRefusesAnActiveAccount(t *testing.T) { + _, nc := newSealTestDaemon(t, &sealTestChecker{terminating: map[string]bool{sealTestTerminating: true}}) + + raw := sealRequest(t, nc, ec2v1.SealAccountSettingsRequest{AccountID: sealTestActive}, "") + assert.Equal(t, awserrors.ErrorIncorrectState, replyErrCode(t, raw)) +} + +// A request without a valid account is refused, not sealed. +func TestSealAccountSettingsSubjectRefusesAMissingAccount(t *testing.T) { + _, nc := newSealTestDaemon(t, &sealTestChecker{terminating: map[string]bool{sealTestTerminating: true}}) + + assert.Equal(t, awserrors.ErrorMissingParameter, replyErrCode(t, sealRequest(t, nc, ec2v1.SealAccountSettingsRequest{}, ""))) +} + +// A daemon that does not serve the subject leaves the request with no responder. +func TestSealAccountSettingsSubjectHasNoResponderWithoutTheDaemon(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + _, err := natsRequestAs(nc, ec2v1.SealAccountSettingsSubject, []byte(`{"account_id":"111122223333"}`), "", time.Second) + require.ErrorIs(t, err, nats.ErrNoResponders) +} diff --git a/spinifex/domains/ec2/account/cleanup.go b/spinifex/domains/ec2/account/cleanup.go deleted file mode 100644 index 183c6086b..000000000 --- a/spinifex/domains/ec2/account/cleanup.go +++ /dev/null @@ -1,90 +0,0 @@ -package account - -import ( - "context" - "errors" - "fmt" - - awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" - awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers" - "github.com/nats-io/nats.go/jetstream" -) - -// AccountSettingsCleanupDisposition is CleanupAccount's outcome for an -// account's settings key. -type AccountSettingsCleanupDisposition string - -const ( - AccountSettingsCleanupDeleted AccountSettingsCleanupDisposition = "deleted" - AccountSettingsCleanupAbsent AccountSettingsCleanupDisposition = "absent" - AccountSettingsCleanupFailed AccountSettingsCleanupDisposition = "failed" -) - -// evidenceRevisionGuardedDelete and evidenceObservedAbsent are the fixed -// evidence bases CleanupAccount reports; see kvstore/bucket.go:112-120 for why -// a plain read cannot support a stronger absence claim. -const ( - evidenceRevisionGuardedDelete = "revision-guarded-delete" - evidenceObservedAbsent = "observed-absent-at-scan" -) - -// cleanupMaxAttempts bounds retries against a revision that keeps moving -// under concurrent writes, so cleanup fails closed instead of spinning. -const cleanupMaxAttempts = 5 - -// AccountSettingsCleanup is CleanupAccount's result: what happened to an -// account's settings key, and the fixed evidence basis for that claim. -type AccountSettingsCleanup struct { - Disposition AccountSettingsCleanupDisposition - Evidence string - Revision uint64 -} - -// CleanupAccount removes accountID's settings key so a reused account ID does -// not inherit a deleted account's EBS-encryption-by-default or -// serial-console-access settings. It is unwired: nothing calls it yet. -func (s *AccountSettingsServiceImpl) CleanupAccount(ctx context.Context, accountID string) (*AccountSettingsCleanup, error) { - if accountID == "" { - return nil, awserrors.Errorf(awserrors.ErrorMissingParameter, "accountID is required for settings cleanup") - } - if !awsidentifiers.IsAccountID(accountID) { - return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "accountID %q is not a valid AWS account ID", accountID) - } - if accountID == awsidentifiers.GlobalAccountID { - return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "settings cleanup must not target the global account") - } - - // settingsKey is never used here: cleanup must act on exactly accountID - // and never fall back to the global key. - for range cleanupMaxAttempts { - entry, err := s.settingsKV.Get(ctx, accountID) - if err != nil { - if errors.Is(err, jetstream.ErrKeyNotFound) { - return &AccountSettingsCleanup{Disposition: AccountSettingsCleanupAbsent, Evidence: evidenceObservedAbsent}, nil - } - return &AccountSettingsCleanup{Disposition: AccountSettingsCleanupFailed}, fmt.Errorf("get account settings for cleanup: %w", err) - } - rev := entry.Revision() - - err = s.settingsKV.Delete(ctx, accountID, jetstream.LastRevision(rev)) - switch { - case err == nil: - return &AccountSettingsCleanup{ - Disposition: AccountSettingsCleanupDeleted, - Evidence: evidenceRevisionGuardedDelete, - Revision: rev, - }, nil - case errors.Is(err, jetstream.ErrKeyNotFound): - return &AccountSettingsCleanup{Disposition: AccountSettingsCleanupAbsent, Evidence: evidenceObservedAbsent}, nil - case errors.Is(err, jetstream.ErrKeyRevisionMismatch): - // A concurrent write landed between the read and the guarded - // delete; retry against whatever revision now holds. - continue - default: - return &AccountSettingsCleanup{Disposition: AccountSettingsCleanupFailed}, fmt.Errorf("delete account settings for cleanup: %w", err) - } - } - - return &AccountSettingsCleanup{Disposition: AccountSettingsCleanupFailed}, - fmt.Errorf("settings cleanup for account: exhausted %d attempts against concurrent writes", cleanupMaxAttempts) -} diff --git a/spinifex/domains/ec2/account/cleanup_test.go b/spinifex/domains/ec2/account/cleanup_test.go deleted file mode 100644 index f81ddc144..000000000 --- a/spinifex/domains/ec2/account/cleanup_test.go +++ /dev/null @@ -1,181 +0,0 @@ -package account - -import ( - "context" - "errors" - "testing" - - "github.com/aws/aws-sdk-go/service/ec2" - awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers" - "github.com/nats-io/nats.go/jetstream" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -// cleanupFakeKV wraps a real KeyValue so a test can inject a race between -// CleanupAccount's read and its guarded delete, or a store failure, -// deterministically rather than through goroutine timing. -type cleanupFakeKV struct { - jetstream.KeyValue - - deleteCalls int - beforeDelete func(attempt int) error - getErr error -} - -func (f *cleanupFakeKV) Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) { - if f.getErr != nil { - return nil, f.getErr - } - return f.KeyValue.Get(ctx, key) -} - -func (f *cleanupFakeKV) Delete(ctx context.Context, key string, opts ...jetstream.KVDeleteOpt) error { - f.deleteCalls++ - if f.beforeDelete != nil { - if err := f.beforeDelete(f.deleteCalls); err != nil { - return err - } - } - return f.KeyValue.Delete(ctx, key, opts...) -} - -func TestCleanupAccount_DeletesTheSettingsKey(t *testing.T) { - svc := setupTestAccountService(t) - ctx := context.Background() - _, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID) - require.NoError(t, err) - - result, err := svc.CleanupAccount(ctx, testAccountID) - require.NoError(t, err) - assert.Equal(t, AccountSettingsCleanupDeleted, result.Disposition) - assert.Equal(t, evidenceRevisionGuardedDelete, result.Evidence) - assert.NotZero(t, result.Revision) - - _, err = svc.settingsKV.Get(ctx, testAccountID) - assert.ErrorIs(t, err, jetstream.ErrKeyNotFound) -} - -func TestCleanupAccount_RepeatCallIsIdempotent(t *testing.T) { - svc := setupTestAccountService(t) - ctx := context.Background() - _, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID) - require.NoError(t, err) - - first, err := svc.CleanupAccount(ctx, testAccountID) - require.NoError(t, err) - assert.Equal(t, AccountSettingsCleanupDeleted, first.Disposition) - - second, err := svc.CleanupAccount(ctx, testAccountID) - require.NoError(t, err) - assert.Equal(t, AccountSettingsCleanupAbsent, second.Disposition) - assert.Equal(t, evidenceObservedAbsent, second.Evidence) -} - -func TestCleanupAccount_AbsentKeyIsIdempotent(t *testing.T) { - svc := setupTestAccountService(t) - result, err := svc.CleanupAccount(context.Background(), "333333333333") - require.NoError(t, err) - assert.Equal(t, AccountSettingsCleanupAbsent, result.Disposition) - assert.Equal(t, evidenceObservedAbsent, result.Evidence) -} - -func TestCleanupAccount_OtherAccountsUntouched(t *testing.T) { - svc := setupTestAccountService(t) - ctx := context.Background() - accountA, accountB := testAccountID, "222222222222" - _, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, accountA) - require.NoError(t, err) - _, err = svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, accountB) - require.NoError(t, err) - - result, err := svc.CleanupAccount(ctx, accountA) - require.NoError(t, err) - assert.Equal(t, AccountSettingsCleanupDeleted, result.Disposition) - - out, err := svc.GetEbsEncryptionByDefault(ctx, &ec2.GetEbsEncryptionByDefaultInput{}, accountB) - require.NoError(t, err) - assert.True(t, *out.EbsEncryptionByDefault) -} - -func TestCleanupAccount_RefusesGlobalAccountID(t *testing.T) { - svc := setupTestAccountService(t) - result, err := svc.CleanupAccount(context.Background(), awsidentifiers.GlobalAccountID) - assert.Error(t, err) - assert.Nil(t, result) -} - -func TestCleanupAccount_RefusesEmptyAccountID(t *testing.T) { - svc := setupTestAccountService(t) - result, err := svc.CleanupAccount(context.Background(), "") - assert.Error(t, err) - assert.Nil(t, result) -} - -func TestCleanupAccount_RefusesInvalidAccountID(t *testing.T) { - svc := setupTestAccountService(t) - result, err := svc.CleanupAccount(context.Background(), "not-an-account") - assert.Error(t, err) - assert.Nil(t, result) -} - -// A concurrent Put landing between CleanupAccount's read and its -// revision-guarded delete must not be lost silently: the delete is rejected -// and retried against the revision the write left behind. -func TestCleanupAccount_ConcurrentPutDuringCleanupRetriesAndDeletes(t *testing.T) { - svc := setupTestAccountService(t) - ctx := context.Background() - _, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID) - require.NoError(t, err) - - fake := &cleanupFakeKV{KeyValue: svc.settingsKV} - fake.beforeDelete = func(attempt int) error { - if attempt != 1 { - return nil - } - // Simulate a racing settings write landing after CleanupAccount's - // read but before its guarded delete, through the real KV so the - // delete's expected revision is now stale. - return svc.saveSettings(ctx, &AccountSettingsRecord{SerialConsoleAccess: true}, testAccountID) - } - svc.settingsKV = fake - - result, err := svc.CleanupAccount(ctx, testAccountID) - require.NoError(t, err) - assert.Equal(t, AccountSettingsCleanupDeleted, result.Disposition) - assert.Equal(t, 2, fake.deleteCalls, "the first delete is rejected by the race and retried once") -} - -// A revision that keeps moving under sustained concurrent writes must not -// spin forever: cleanup gives up and reports failed, bounded by attempts. -func TestCleanupAccount_PersistentConcurrentWritesFailsBounded(t *testing.T) { - svc := setupTestAccountService(t) - ctx := context.Background() - _, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID) - require.NoError(t, err) - - fake := &cleanupFakeKV{KeyValue: svc.settingsKV} - fake.beforeDelete = func(int) error { - return svc.saveSettings(ctx, &AccountSettingsRecord{SerialConsoleAccess: true}, testAccountID) - } - svc.settingsKV = fake - - result, err := svc.CleanupAccount(ctx, testAccountID) - assert.Error(t, err) - require.NotNil(t, result) - assert.Equal(t, AccountSettingsCleanupFailed, result.Disposition) - assert.Equal(t, cleanupMaxAttempts, fake.deleteCalls) -} - -// A store error must report failed, never absent: absence is a claim about -// what the store holds, which an error gives no basis to make. -func TestCleanupAccount_StoreErrorIsFailedNotAbsent(t *testing.T) { - svc := setupTestAccountService(t) - fake := &cleanupFakeKV{KeyValue: svc.settingsKV, getErr: errors.New("kv unavailable")} - svc.settingsKV = fake - - result, err := svc.CleanupAccount(context.Background(), testAccountID) - assert.Error(t, err) - require.NotNil(t, result) - assert.Equal(t, AccountSettingsCleanupFailed, result.Disposition) -} diff --git a/spinifex/domains/ec2/account/seal.go b/spinifex/domains/ec2/account/seal.go new file mode 100644 index 000000000..ad8e67ac3 --- /dev/null +++ b/spinifex/domains/ec2/account/seal.go @@ -0,0 +1,135 @@ +package account + +import ( + "context" + "encoding/json" + "errors" + "log/slog" + + awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" + awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers" + "github.com/nats-io/nats.go/jetstream" +) + +// sealMaxAttempts bounds SealAccount's retry loop against a revision that +// keeps moving under concurrent writes, so it fails closed instead of +// spinning. +const sealMaxAttempts = 5 + +// sealedRecord is the fixed value a seal writes. Clearing both booleans +// retains no tenant setting; an older reader that ignores state sees the +// AWS defaults. +var sealedRecord = &AccountSettingsRecord{State: AccountSettingsStateSealed} + +// AccountSettingsSeal is SealAccount's result: the revision at which the +// stream leader accepted the sealed value. +type AccountSettingsSeal struct { + Revision uint64 +} + +// AccountTerminationChecker reports whether IAM holds accountID as TERMINATING. +// A missing record is (false, nil); only a failed read is an error. +type AccountTerminationChecker interface { + AccountTerminating(ctx context.Context, accountID string) (bool, error) +} + +// SetAccountTerminationChecker wires the precondition; until it is set, SealAccount refuses. +func (s *AccountSettingsServiceImpl) SetAccountTerminationChecker(c AccountTerminationChecker) { + s.terminationChecker = c +} + +// checkTerminating runs before any settings-KV call, so a refusal never writes. +// Not TERMINATING, including a missing IAM record, is IncorrectState; a read +// failure or an unwired checker is InternalError. +func (s *AccountSettingsServiceImpl) checkTerminating(ctx context.Context, accountID string) error { + if s.terminationChecker == nil { + slog.ErrorContext(ctx, "Account settings seal has no termination checker wired", "accountID", accountID) + return errors.New(awserrors.ErrorInternalError) + } + + terminating, err := s.terminationChecker.AccountTerminating(ctx, accountID) + if err != nil { + slog.ErrorContext(ctx, "Failed to read account status for seal precondition", "accountID", accountID, "err", err) + return errors.New(awserrors.ErrorInternalError) + } + if !terminating { + return awserrors.Errorf(awserrors.ErrorIncorrectState, "account %s is not TERMINATING", accountID) + } + return nil +} + +// isRevisionConflict reports a lost CAS race SealAccount should retry. Besides +// the two sentinels, Create's retry against a delete marker returns a bare +// *jetstream.APIError carrying a wrong-last-sequence code. +func isRevisionConflict(err error) bool { + if errors.Is(err, jetstream.ErrKeyExists) || errors.Is(err, jetstream.ErrKeyRevisionMismatch) { + return true + } + if apiErr, ok := errors.AsType[*jetstream.APIError](err); ok { + return apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence || + apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequenceConstant + } + return false +} + +// SealAccount seals accountID's settings record so no later mutation can +// recreate it. It never relies on a read for its claim: the only success is a +// leader-accepted write, evidenced by the returned revision. +func (s *AccountSettingsServiceImpl) SealAccount(ctx context.Context, accountID string) (*AccountSettingsSeal, error) { + if accountID == "" { + return nil, awserrors.Errorf(awserrors.ErrorMissingParameter, "accountID is required to seal settings") + } + if !awsidentifiers.IsAccountID(accountID) { + return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "accountID %q is not a valid AWS account ID", accountID) + } + if accountID == awsidentifiers.GlobalAccountID { + return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "settings seal must not target the global account") + } + + if err := s.checkTerminating(ctx, accountID); err != nil { + return nil, err + } + + data, err := json.Marshal(sealedRecord) + if err != nil { + slog.ErrorContext(ctx, "Failed to encode sealed account settings record", "accountID", accountID, "err", err) + return nil, errors.New(awserrors.ErrorInternalError) + } + + // settingsKey is never used here: a seal must act on exactly accountID + // and never fall back to the global key. + for range sealMaxAttempts { + var rev uint64 + entry, getErr := s.settingsKV.Get(ctx, accountID) + switch { + case getErr == nil: + rev = entry.Revision() + case errors.Is(getErr, jetstream.ErrKeyNotFound): + rev = 0 + default: + slog.ErrorContext(ctx, "Failed to read account settings for seal", "accountID", accountID, "err", getErr) + return nil, errors.New(awserrors.ErrorInternalError) + } + + var newRev uint64 + var writeErr error + if rev == 0 { + newRev, writeErr = s.settingsKV.Create(ctx, accountID, data) + } else { + newRev, writeErr = s.settingsKV.Update(ctx, accountID, data, rev) + } + + switch { + case writeErr == nil: + return &AccountSettingsSeal{Revision: newRev}, nil + case isRevisionConflict(writeErr): + continue + default: + slog.ErrorContext(ctx, "Failed to write sealed account settings record", "accountID", accountID, "err", writeErr) + return nil, errors.New(awserrors.ErrorInternalError) + } + } + + slog.ErrorContext(ctx, "Account settings seal exhausted attempts against concurrent writes", "accountID", accountID, "attempts", sealMaxAttempts) + return nil, errors.New(awserrors.ErrorInternalError) +} diff --git a/spinifex/domains/ec2/account/seal_test.go b/spinifex/domains/ec2/account/seal_test.go new file mode 100644 index 000000000..76584d704 --- /dev/null +++ b/spinifex/domains/ec2/account/seal_test.go @@ -0,0 +1,559 @@ +package account + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sync" + "testing" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/internal/testkit" + awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" + awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers" + "github.com/nats-io/nats.go/jetstream" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// stubTerminationChecker answers AccountTerminating from a fixed set, for +// tests that exercise the KV paths rather than IAM itself. +type stubTerminationChecker struct { + terminating map[string]bool + err error +} + +func (c *stubTerminationChecker) AccountTerminating(_ context.Context, accountID string) (bool, error) { + if c.err != nil { + return false, c.err + } + return c.terminating[accountID], nil +} + +// terminatingChecker answers true for exactly accountID. +func terminatingChecker(accountID string) *stubTerminationChecker { + return &stubTerminationChecker{terminating: map[string]bool{accountID: true}} +} + +// lastSettingsSeq reports the account-settings stream's last sequence, so a +// test can assert a failed attempt wrote nothing. +func lastSettingsSeq(t *testing.T, svc *AccountSettingsServiceImpl) uint64 { + t.Helper() + stream, err := svc.js.Stream(context.Background(), "KV_"+KVBucketAccountSettings) + require.NoError(t, err) + info, err := stream.Info(context.Background()) + require.NoError(t, err) + return info.State.LastSeq +} + +// mutationCase is one of the four settings mutations, called uniformly so a +// test can table over them. +type mutationCase struct { + name string + call func(svc *AccountSettingsServiceImpl, ctx context.Context) error +} + +func mutationTable() []mutationCase { + return []mutationCase{ + {"EnableEbs", func(svc *AccountSettingsServiceImpl, ctx context.Context) error { + _, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID) + return err + }}, + {"DisableEbs", func(svc *AccountSettingsServiceImpl, ctx context.Context) error { + _, err := svc.DisableEbsEncryptionByDefault(ctx, &ec2.DisableEbsEncryptionByDefaultInput{}, testAccountID) + return err + }}, + {"EnableSerial", func(svc *AccountSettingsServiceImpl, ctx context.Context) error { + _, err := svc.EnableSerialConsoleAccess(ctx, &ec2.EnableSerialConsoleAccessInput{}, testAccountID) + return err + }}, + {"DisableSerial", func(svc *AccountSettingsServiceImpl, ctx context.Context) error { + _, err := svc.DisableSerialConsoleAccess(ctx, &ec2.DisableSerialConsoleAccessInput{}, testAccountID) + return err + }}, + } +} + +// seedSettingsState prepares testAccountID's record for a race test: present +// (an active record), absent (never written) or tombstone (written then +// deleted, leaving a delete marker Get hides). +func seedSettingsState(t *testing.T, svc *AccountSettingsServiceImpl, state string) { + t.Helper() + ctx := context.Background() + switch state { + case "present": + _, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID) + require.NoError(t, err) + case "tombstone": + _, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID) + require.NoError(t, err) + require.NoError(t, svc.settingsKV.Delete(ctx, testAccountID)) + case "absent": + default: + t.Fatalf("unknown seed state %q", state) + } +} + +// sealRaceKV wraps a real KeyValue so a test can inject a deterministic race +// between a caller's read and its guarded write (Create or Update), instead +// of relying on goroutine timing. +type sealRaceKV struct { + jetstream.KeyValue + + writeCalls int + beforeWrite func(attempt int) error +} + +func (f *sealRaceKV) Create(ctx context.Context, key string, value []byte, opts ...jetstream.KVCreateOpt) (uint64, error) { + f.writeCalls++ + if f.beforeWrite != nil { + if err := f.beforeWrite(f.writeCalls); err != nil { + return 0, err + } + } + return f.KeyValue.Create(ctx, key, value, opts...) +} + +func (f *sealRaceKV) Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) { + f.writeCalls++ + if f.beforeWrite != nil { + if err := f.beforeWrite(f.writeCalls); err != nil { + return 0, err + } + } + return f.KeyValue.Update(ctx, key, value, revision) +} + +// staleGetKV always answers Get with a fixed snapshot taken at construction, +// so a caller's CAS keeps retrying against a revision that can never win. +type staleGetKV struct { + jetstream.KeyValue + + entry jetstream.KeyValueEntry +} + +func (f *staleGetKV) Get(context.Context, string) (jetstream.KeyValueEntry, error) { + return f.entry, nil +} + +// staleNotFoundKV always answers Get as absent, regardless of what the real +// bucket holds. +type staleNotFoundKV struct { + jetstream.KeyValue +} + +func (f *staleNotFoundKV) Get(context.Context, string) (jetstream.KeyValueEntry, error) { + return nil, jetstream.ErrKeyNotFound +} + +// errGetKV always fails Get with a fixed store error. +type errGetKV struct { + jetstream.KeyValue + + err error +} + +func (f *errGetKV) Get(context.Context, string) (jetstream.KeyValueEntry, error) { + return nil, f.err +} + +func decodeSettings(t *testing.T, entry jetstream.KeyValueEntry) AccountSettingsRecord { + t.Helper() + var rec AccountSettingsRecord + require.NoError(t, json.Unmarshal(entry.Value(), &rec)) + return rec +} + +// A seal landing between a mutation's read and its guarded write is not +// outlived: the mutation's retry sees the seal and is rejected, and nothing +// it wrote survives. +func TestSealRaceMutation_SealLandsBetweenReadAndWrite(t *testing.T) { + for _, m := range mutationTable() { + for _, state := range []string{"present", "absent", "tombstone"} { + t.Run(m.name+"_"+state, func(t *testing.T) { + svc := setupTestAccountService(t) + ctx := context.Background() + seedSettingsState(t, svc, state) + + realKV := svc.settingsKV + sealer := &AccountSettingsServiceImpl{js: svc.js, settingsKV: realKV} + sealer.SetAccountTerminationChecker(terminatingChecker(testAccountID)) + + var sealed *AccountSettingsSeal + race := &sealRaceKV{KeyValue: realKV} + race.beforeWrite = func(attempt int) error { + if attempt != 1 { + return nil + } + var sealErr error + sealed, sealErr = sealer.SealAccount(ctx, testAccountID) + return sealErr + } + svc.settingsKV = race + + err := m.call(svc, ctx) + require.Error(t, err) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked), "got %v", err) + require.NotNil(t, sealed) + + entry, err := realKV.Get(ctx, testAccountID) + require.NoError(t, err) + assert.Equal(t, AccountSettingsStateSealed, decodeSettings(t, entry).State) + assert.Equal(t, sealed.Revision, entry.Revision(), "no write landed after the seal") + }) + } + } +} + +// A mutation landing inside a seal's read-write window does not survive it: +// the seal retries against the mutation's write and wins, leaving the +// sealed value with both booleans false. +func TestSealRaceMutation_MutationLandsBetweenSealReadAndWrite(t *testing.T) { + for _, m := range mutationTable() { + for _, state := range []string{"present", "absent", "tombstone"} { + t.Run(m.name+"_"+state, func(t *testing.T) { + svc := setupTestAccountService(t) + ctx := context.Background() + seedSettingsState(t, svc, state) + + realKV := svc.settingsKV + race := &sealRaceKV{KeyValue: realKV} + race.beforeWrite = func(attempt int) error { + if attempt != 1 { + return nil + } + _ = m.call(svc, ctx) + return nil + } + + sealer := &AccountSettingsServiceImpl{js: svc.js, settingsKV: race} + sealer.SetAccountTerminationChecker(terminatingChecker(testAccountID)) + + seal, err := sealer.SealAccount(ctx, testAccountID) + require.NoError(t, err) + + entry, err := realKV.Get(ctx, testAccountID) + require.NoError(t, err) + rec := decodeSettings(t, entry) + assert.Equal(t, AccountSettingsStateSealed, rec.State) + assert.False(t, rec.EbsEncryptionByDefault) + assert.False(t, rec.SerialConsoleAccess) + assert.Equal(t, seal.Revision, entry.Revision()) + }) + } + } +} + +// None of the four mutations can act on an already-sealed record. +func TestMutationOnSealedRecordIsRejected(t *testing.T) { + for _, m := range mutationTable() { + t.Run(m.name, func(t *testing.T) { + svc := setupTestAccountService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(terminatingChecker(testAccountID)) + _, err := svc.SealAccount(ctx, testAccountID) + require.NoError(t, err) + + before, err := svc.settingsKV.Get(ctx, testAccountID) + require.NoError(t, err) + + err = m.call(svc, ctx) + require.Error(t, err) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked), "got %v", err) + + after, err := svc.settingsKV.Get(ctx, testAccountID) + require.NoError(t, err) + assert.Equal(t, before.Revision(), after.Revision()) + }) + } +} + +// Every settings mutation races one seal on a real bucket; afterwards the +// record is sealed and stays sealed, so no success among them could have +// landed after the seal. +func TestSealConcurrentMutations(t *testing.T) { + svc := setupTestAccountService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(terminatingChecker(testAccountID)) + _, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID) + require.NoError(t, err) + + mutations := mutationTable() + errs := make([]error, len(mutations)) + var wg sync.WaitGroup + for i, m := range mutations { + wg.Add(1) + go func(i int, m mutationCase) { + defer wg.Done() + errs[i] = m.call(svc, ctx) + }(i, m) + } + + var sealResult *AccountSettingsSeal + var sealErr error + wg.Go(func() { + sealResult, sealErr = svc.SealAccount(ctx, testAccountID) + }) + wg.Wait() + + require.NoError(t, sealErr) + require.NotNil(t, sealResult) + for i, err := range errs { + if err != nil { + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked), "mutation %s: got %v", mutations[i].name, err) + } + } + + entry, err := svc.settingsKV.Get(ctx, testAccountID) + require.NoError(t, err) + assert.Equal(t, AccountSettingsStateSealed, decodeSettings(t, entry).State) + assert.Equal(t, sealResult.Revision, entry.Revision()) + + _, err = svc.GetEbsEncryptionByDefault(ctx, &ec2.GetEbsEncryptionByDefaultInput{}, testAccountID) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked)) +} + +// A Get pinned to a revision the record has already moved past exhausts the +// mutation's retries without ever writing. +func TestMutationStaleReadExhaustsWithoutWriting(t *testing.T) { + svc := setupTestAccountService(t) + ctx := context.Background() + + _, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID) + require.NoError(t, err) + stale, err := svc.settingsKV.Get(ctx, testAccountID) + require.NoError(t, err) + + _, err = svc.DisableEbsEncryptionByDefault(ctx, &ec2.DisableEbsEncryptionByDefaultInput{}, testAccountID) + require.NoError(t, err) + + before := lastSettingsSeq(t, svc) + svc.settingsKV = &staleGetKV{KeyValue: svc.settingsKV, entry: stale} + + _, err = svc.EnableSerialConsoleAccess(ctx, &ec2.EnableSerialConsoleAccessInput{}, testAccountID) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err) + assert.Equal(t, before, lastSettingsSeq(t, svc)) +} + +// A Get that always reports absent, while a sealed record actually exists, +// never gets to recreate it: every Create attempt conflicts. +func TestMutationStaleAbsentDoesNotRecreate(t *testing.T) { + svc := setupTestAccountService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(terminatingChecker(testAccountID)) + _, err := svc.SealAccount(ctx, testAccountID) + require.NoError(t, err) + + before := lastSettingsSeq(t, svc) + svc.settingsKV = &staleNotFoundKV{KeyValue: svc.settingsKV} + + _, err = svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err) + assert.Equal(t, before, lastSettingsSeq(t, svc)) +} + +// A revision that keeps moving under sustained concurrent writes must not +// let the seal spin forever: it gives up, bounded by its attempt budget, +// and its own write never lands. +func TestSealSustainedConflictFailsWithoutWriting(t *testing.T) { + svc := setupTestAccountService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(terminatingChecker(testAccountID)) + _, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID) + require.NoError(t, err) + + realKV := svc.settingsKV + race := &sealRaceKV{KeyValue: realKV} + race.beforeWrite = func(int) error { + data, merr := json.Marshal(&AccountSettingsRecord{SerialConsoleAccess: true}) + require.NoError(t, merr) + _, perr := realKV.Put(ctx, testAccountID, data) + return perr + } + svc.settingsKV = race + + _, err = svc.SealAccount(ctx, testAccountID) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err) + assert.Equal(t, sealMaxAttempts, race.writeCalls) + + entry, err := realKV.Get(ctx, testAccountID) + require.NoError(t, err) + assert.NotEqual(t, AccountSettingsStateSealed, decodeSettings(t, entry).State, "the seal's own write never landed") +} + +// A store error on the seal's read reports failed, never written. +func TestSealStoreErrorWritesNothing(t *testing.T) { + svc := setupTestAccountService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(terminatingChecker(testAccountID)) + + before := lastSettingsSeq(t, svc) + svc.settingsKV = &errGetKV{KeyValue: svc.settingsKV, err: errors.New("kv unavailable")} + + _, err := svc.SealAccount(ctx, testAccountID) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err) + assert.Equal(t, before, lastSettingsSeq(t, svc)) +} + +func TestIsRevisionConflictRecognisesTombstoneRetry(t *testing.T) { + assert.True(t, isRevisionConflict(jetstream.ErrKeyExists)) + assert.True(t, isRevisionConflict(jetstream.ErrKeyRevisionMismatch)) + assert.True(t, isRevisionConflict(&jetstream.APIError{ErrorCode: jetstream.JSErrCodeStreamWrongLastSequence})) + assert.True(t, isRevisionConflict(&jetstream.APIError{ErrorCode: jetstream.JSErrCodeStreamWrongLastSequenceConstant})) + assert.True(t, isRevisionConflict(fmt.Errorf("wrapped: %w", &jetstream.APIError{ErrorCode: jetstream.JSErrCodeStreamWrongLastSequence}))) + assert.False(t, isRevisionConflict(errors.New("unrelated failure"))) + assert.False(t, isRevisionConflict(nil)) +} + +// An account IAM does not hold as TERMINATING refuses the seal before any +// settings-KV write, whether the key is present or absent. +func TestSealRefusedForActiveAccount(t *testing.T) { + for _, state := range []string{"present", "absent"} { + t.Run(state, func(t *testing.T) { + svc := setupTestAccountService(t) + ctx := context.Background() + if state == "present" { + _, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID) + require.NoError(t, err) + } + svc.SetAccountTerminationChecker(&stubTerminationChecker{terminating: map[string]bool{}}) + + before := lastSettingsSeq(t, svc) + _, err := svc.SealAccount(ctx, testAccountID) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorIncorrectState), "got %v", err) + assert.Equal(t, before, lastSettingsSeq(t, svc)) + }) + } +} + +func TestSealRefusedForMissingIAMRecord(t *testing.T) { + svc := setupTestAccountService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(&stubTerminationChecker{terminating: map[string]bool{}}) + + before := lastSettingsSeq(t, svc) + _, err := svc.SealAccount(ctx, testAccountID) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorIncorrectState), "got %v", err) + assert.Equal(t, before, lastSettingsSeq(t, svc)) +} + +func TestSealRefusedWhenIAMReadFails(t *testing.T) { + svc := setupTestAccountService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(&stubTerminationChecker{err: errors.New("iam unavailable")}) + + before := lastSettingsSeq(t, svc) + _, err := svc.SealAccount(ctx, testAccountID) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err) + assert.Equal(t, before, lastSettingsSeq(t, svc)) +} + +func TestSealRefusedWithNoCheckerWired(t *testing.T) { + svc := setupTestAccountService(t) + ctx := context.Background() + + before := lastSettingsSeq(t, svc) + _, err := svc.SealAccount(ctx, testAccountID) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err) + assert.Equal(t, before, lastSettingsSeq(t, svc)) +} + +func TestSealAcceptedForTerminatingAccount(t *testing.T) { + svc := setupTestAccountService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(terminatingChecker(testAccountID)) + + seal, err := svc.SealAccount(ctx, testAccountID) + require.NoError(t, err) + assert.NotZero(t, seal.Revision) + + entry, err := svc.settingsKV.Get(ctx, testAccountID) + require.NoError(t, err) + assert.Equal(t, seal.Revision, entry.Revision()) +} + +// A sealed record stays sealed after the service is rebuilt over the same +// JetStream, and every one of the six settings actions still refuses it. +func TestSealSurvivesServiceReconstruction(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + ctx := context.Background() + + svc, err := NewAccountSettingsServiceImplWithNATS(ctx, nil, nc) + require.NoError(t, err) + svc.SetAccountTerminationChecker(terminatingChecker(testAccountID)) + first, err := svc.SealAccount(ctx, testAccountID) + require.NoError(t, err) + + rebuilt, err := NewAccountSettingsServiceImplWithNATS(ctx, nil, nc) + require.NoError(t, err) + + entry, err := rebuilt.settingsKV.Get(ctx, testAccountID) + require.NoError(t, err) + assert.Equal(t, AccountSettingsStateSealed, decodeSettings(t, entry).State) + + _, err = rebuilt.GetEbsEncryptionByDefault(ctx, &ec2.GetEbsEncryptionByDefaultInput{}, testAccountID) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked)) + _, err = rebuilt.GetSerialConsoleAccessStatus(ctx, &ec2.GetSerialConsoleAccessStatusInput{}, testAccountID) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked)) + _, err = rebuilt.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked)) + _, err = rebuilt.DisableEbsEncryptionByDefault(ctx, &ec2.DisableEbsEncryptionByDefaultInput{}, testAccountID) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked)) + _, err = rebuilt.EnableSerialConsoleAccess(ctx, &ec2.EnableSerialConsoleAccessInput{}, testAccountID) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked)) + _, err = rebuilt.DisableSerialConsoleAccess(ctx, &ec2.DisableSerialConsoleAccessInput{}, testAccountID) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorBlocked)) + + rebuilt.SetAccountTerminationChecker(terminatingChecker(testAccountID)) + second, err := rebuilt.SealAccount(ctx, testAccountID) + require.NoError(t, err) + assert.Greater(t, second.Revision, first.Revision) +} + +func TestSealAbsentKeyCreatesSealedRecord(t *testing.T) { + svc := setupTestAccountService(t) + ctx := context.Background() + svc.SetAccountTerminationChecker(terminatingChecker(testAccountID)) + + _, err := svc.settingsKV.Get(ctx, testAccountID) + require.ErrorIs(t, err, jetstream.ErrKeyNotFound) + + seal, err := svc.SealAccount(ctx, testAccountID) + require.NoError(t, err) + + entry, err := svc.settingsKV.Get(ctx, testAccountID) + require.NoError(t, err) + assert.Equal(t, seal.Revision, entry.Revision()) + assert.Equal(t, AccountSettingsStateSealed, decodeSettings(t, entry).State) +} + +func TestSealRefusesEmptyInvalidAndGlobalAccount(t *testing.T) { + svc := setupTestAccountService(t) + ctx := context.Background() + + _, err := svc.SealAccount(ctx, "") + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorMissingParameter), "got %v", err) + + _, err = svc.SealAccount(ctx, "not-an-account") + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInvalidParameterValue), "got %v", err) + + _, err = svc.SealAccount(ctx, awsidentifiers.GlobalAccountID) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInvalidParameterValue), "got %v", err) +} + +func TestSealLeavesOtherAccountsUntouched(t *testing.T) { + svc := setupTestAccountService(t) + ctx := context.Background() + accountB := "222222222222" + _, err := svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, accountB) + require.NoError(t, err) + + svc.SetAccountTerminationChecker(terminatingChecker(testAccountID)) + _, err = svc.SealAccount(ctx, testAccountID) + require.NoError(t, err) + + out, err := svc.GetEbsEncryptionByDefault(ctx, &ec2.GetEbsEncryptionByDefaultInput{}, accountB) + require.NoError(t, err) + assert.True(t, *out.EbsEncryptionByDefault) +} diff --git a/spinifex/domains/ec2/account/service_impl.go b/spinifex/domains/ec2/account/service_impl.go index eb4bfb03b..19c6b3579 100644 --- a/spinifex/domains/ec2/account/service_impl.go +++ b/spinifex/domains/ec2/account/service_impl.go @@ -5,12 +5,13 @@ import ( "encoding/json" "errors" "fmt" - awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers" "log/slog" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/spinifex/bootstrap/config" + awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" + awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" "github.com/nats-io/nats.go" @@ -20,14 +21,19 @@ import ( const ( KVBucketAccountSettings = "spinifex-ec2-account-settings" KVBucketAccountSettingsVersion = 1 - KeyEbsEncryptionDefault = "ebs-encryption-default" - KeySerialConsoleAccess = "serial-console-access" ) +// AccountSettingsState is a settings record's lifecycle state. "" is active; +// a record written before the state field existed has none and stays active. +type AccountSettingsState string + +const AccountSettingsStateSealed AccountSettingsState = "sealed" + // AccountSettingsRecord represents stored account settings. type AccountSettingsRecord struct { - EbsEncryptionByDefault bool `json:"ebs_encryption_by_default"` - SerialConsoleAccess bool `json:"serial_console_access"` + EbsEncryptionByDefault bool `json:"ebs_encryption_by_default"` + SerialConsoleAccess bool `json:"serial_console_access"` + State AccountSettingsState `json:"state,omitempty"` } // AccountSettingsServiceImpl implements account settings operations with NATS JetStream persistence. @@ -35,6 +41,10 @@ type AccountSettingsServiceImpl struct { config *config.Config js jetstream.JetStream settingsKV jetstream.KeyValue + + // terminationChecker is the seal precondition; SealAccount refuses until + // it is wired. + terminationChecker AccountTerminationChecker } var _ AccountSettingsService = (*AccountSettingsServiceImpl)(nil) @@ -72,50 +82,99 @@ func settingsKey(accountID string) string { return accountID } -// getSettings retrieves current account settings. +// errSettingsSealed marks a mutation or read that found a sealed record, so +// callers map it to Blocked without logging it as a fault. +var errSettingsSealed = errors.New(awserrors.ErrorBlocked) + +// errSettingsContended marks a mutation whose CAS retries were exhausted +// under concurrent writers, bounding the retry loop instead of spinning. +var errSettingsContended = errors.New("account settings: CAS retries exhausted") + +// checkActive rejects a sealed or unknown-state record; "" is active and +// passes through. A mutation must never act on a record that failed this. +func checkActive(rec *AccountSettingsRecord) error { + switch rec.State { + case "": + return nil + case AccountSettingsStateSealed: + return errSettingsSealed + default: + return fmt.Errorf("account settings: unknown state %q", rec.State) + } +} + +// mapSettingsError turns a mutateSettings or getSettings failure into the AWS +// error a caller returns: sealed is an expected, unlogged Blocked; everything +// else is InternalError, logged with the account ID. +func mapSettingsError(ctx context.Context, accountID string, err error) error { + switch { + case err == nil: + return nil + case errors.Is(err, errSettingsSealed): + return errors.New(awserrors.ErrorBlocked) + case errors.Is(err, errSettingsContended): + slog.ErrorContext(ctx, "Account settings CAS retries exhausted under contention", "accountID", accountID) + case errors.Is(err, kvutil.ErrRead): + slog.ErrorContext(ctx, "Failed to read account settings from KV", "accountID", accountID, "err", err) + case errors.Is(err, kvutil.ErrDecode): + slog.ErrorContext(ctx, "Corrupt account settings record in KV", "accountID", accountID, "err", err) + case errors.Is(err, kvutil.ErrEncode): + slog.ErrorContext(ctx, "Failed to marshal account settings record", "accountID", accountID, "err", err) + case errors.Is(err, kvutil.ErrWrite): + slog.ErrorContext(ctx, "Failed to write account settings to KV", "accountID", accountID, "err", err) + default: + slog.ErrorContext(ctx, "Account settings operation failed", "accountID", accountID, "err", err) + } + return errors.New(awserrors.ErrorInternalError) +} + +// mutateSettings applies set to accountID's record at the revision it read. +// changed is always true, so every success is a leader-accepted write over an +// active value; a seal landing between the read and the write fails the CAS. +func (s *AccountSettingsServiceImpl) mutateSettings(ctx context.Context, accountID string, set func(*AccountSettingsRecord)) error { + _, err := kvutil.Update(ctx, s.settingsKV, settingsKey(accountID), kvutil.CASConfig{ + CreateIfAbsent: true, + Exhausted: func(string, int) error { return errSettingsContended }, + }, func(rec *AccountSettingsRecord) (bool, error) { + if err := checkActive(rec); err != nil { + return false, err + } + set(rec) + return true, nil + }) + return mapSettingsError(ctx, accountID, err) +} + +// getSettings retrieves current account settings, rejecting a sealed or +// unknown-state record so a read never answers from behind a seal. Absent is +// the AWS defaults; a read is a direct get and can be stale. func (s *AccountSettingsServiceImpl) getSettings(ctx context.Context, accountID string) (*AccountSettingsRecord, error) { entry, err := s.settingsKV.Get(ctx, settingsKey(accountID)) if err != nil { if errors.Is(err, jetstream.ErrKeyNotFound) { - return &AccountSettingsRecord{ - EbsEncryptionByDefault: false, - SerialConsoleAccess: false, - }, nil + return &AccountSettingsRecord{}, nil } - return nil, fmt.Errorf("failed to get account settings: %w", err) + slog.ErrorContext(ctx, "Failed to read account settings from KV", "accountID", accountID, "err", err) + return nil, errors.New(awserrors.ErrorInternalError) } var settings AccountSettingsRecord if err := json.Unmarshal(entry.Value(), &settings); err != nil { - return nil, err + slog.ErrorContext(ctx, "Corrupt account settings record in KV", "accountID", accountID, "err", err) + return nil, errors.New(awserrors.ErrorInternalError) + } + if err := checkActive(&settings); err != nil { + return nil, mapSettingsError(ctx, accountID, err) } return &settings, nil } -// saveSettings saves current account settings. -func (s *AccountSettingsServiceImpl) saveSettings(ctx context.Context, settings *AccountSettingsRecord, accountID string) error { - data, err := json.Marshal(settings) - if err != nil { - return err - } - - _, err = s.settingsKV.Put(ctx, settingsKey(accountID), data) - return err -} - // EnableEbsEncryptionByDefault enables EBS encryption by default for the account. func (s *AccountSettingsServiceImpl) EnableEbsEncryptionByDefault(ctx context.Context, input *ec2.EnableEbsEncryptionByDefaultInput, accountID string) (*ec2.EnableEbsEncryptionByDefaultOutput, error) { slog.InfoContext(ctx, "EnableEbsEncryptionByDefault called", "accountID", accountID) - settings, err := s.getSettings(ctx, accountID) - if err != nil { - return nil, err - } - - settings.EbsEncryptionByDefault = true - - if err := s.saveSettings(ctx, settings, accountID); err != nil { + if err := s.mutateSettings(ctx, accountID, func(rec *AccountSettingsRecord) { rec.EbsEncryptionByDefault = true }); err != nil { return nil, err } @@ -128,14 +187,7 @@ func (s *AccountSettingsServiceImpl) EnableEbsEncryptionByDefault(ctx context.Co func (s *AccountSettingsServiceImpl) DisableEbsEncryptionByDefault(ctx context.Context, input *ec2.DisableEbsEncryptionByDefaultInput, accountID string) (*ec2.DisableEbsEncryptionByDefaultOutput, error) { slog.InfoContext(ctx, "DisableEbsEncryptionByDefault called", "accountID", accountID) - settings, err := s.getSettings(ctx, accountID) - if err != nil { - return nil, err - } - - settings.EbsEncryptionByDefault = false - - if err := s.saveSettings(ctx, settings, accountID); err != nil { + if err := s.mutateSettings(ctx, accountID, func(rec *AccountSettingsRecord) { rec.EbsEncryptionByDefault = false }); err != nil { return nil, err } @@ -176,14 +228,7 @@ func (s *AccountSettingsServiceImpl) GetSerialConsoleAccessStatus(ctx context.Co func (s *AccountSettingsServiceImpl) EnableSerialConsoleAccess(ctx context.Context, input *ec2.EnableSerialConsoleAccessInput, accountID string) (*ec2.EnableSerialConsoleAccessOutput, error) { slog.InfoContext(ctx, "EnableSerialConsoleAccess called", "accountID", accountID) - settings, err := s.getSettings(ctx, accountID) - if err != nil { - return nil, err - } - - settings.SerialConsoleAccess = true - - if err := s.saveSettings(ctx, settings, accountID); err != nil { + if err := s.mutateSettings(ctx, accountID, func(rec *AccountSettingsRecord) { rec.SerialConsoleAccess = true }); err != nil { return nil, err } @@ -196,14 +241,7 @@ func (s *AccountSettingsServiceImpl) EnableSerialConsoleAccess(ctx context.Conte func (s *AccountSettingsServiceImpl) DisableSerialConsoleAccess(ctx context.Context, input *ec2.DisableSerialConsoleAccessInput, accountID string) (*ec2.DisableSerialConsoleAccessOutput, error) { slog.InfoContext(ctx, "DisableSerialConsoleAccess called", "accountID", accountID) - settings, err := s.getSettings(ctx, accountID) - if err != nil { - return nil, err - } - - settings.SerialConsoleAccess = false - - if err := s.saveSettings(ctx, settings, accountID); err != nil { + if err := s.mutateSettings(ctx, accountID, func(rec *AccountSettingsRecord) { rec.SerialConsoleAccess = false }); err != nil { return nil, err } diff --git a/spinifex/domains/ec2/account/service_impl_test.go b/spinifex/domains/ec2/account/service_impl_test.go index adc47371c..44209f7b8 100644 --- a/spinifex/domains/ec2/account/service_impl_test.go +++ b/spinifex/domains/ec2/account/service_impl_test.go @@ -2,11 +2,13 @@ package account import ( "context" + "encoding/json" awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers" "testing" "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/internal/testkit" + awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -143,3 +145,44 @@ func TestSettingsKey_EmptyAccountIDFallback(t *testing.T) { assert.Equal(t, awsidentifiers.GlobalAccountID, settingsKey("")) assert.Equal(t, "123456789012", settingsKey("123456789012")) } + +// A raw pre-seal record, with no state field at all, is readable and +// mutable, and a mutation writes it back without adding one. +func TestRecordWithoutStateIsActive(t *testing.T) { + svc := setupTestAccountService(t) + ctx := context.Background() + _, err := svc.settingsKV.Put(ctx, testAccountID, []byte(`{"ebs_encryption_by_default":true,"serial_console_access":false}`)) + require.NoError(t, err) + + out, err := svc.GetEbsEncryptionByDefault(ctx, &ec2.GetEbsEncryptionByDefaultInput{}, testAccountID) + require.NoError(t, err) + assert.True(t, *out.EbsEncryptionByDefault) + + _, err = svc.EnableSerialConsoleAccess(ctx, &ec2.EnableSerialConsoleAccessInput{}, testAccountID) + require.NoError(t, err) + + entry, err := svc.settingsKV.Get(ctx, testAccountID) + require.NoError(t, err) + assert.NotContains(t, string(entry.Value()), "state") +} + +// A record with an unrecognised state fails closed rather than being treated +// as active or sealed, for both a read and a mutation. +func TestUnknownStateFailsClosed(t *testing.T) { + svc := setupTestAccountService(t) + ctx := context.Background() + _, err := svc.settingsKV.Put(ctx, testAccountID, []byte(`{"ebs_encryption_by_default":false,"serial_console_access":false,"state":"quarantined"}`)) + require.NoError(t, err) + + _, err = svc.GetEbsEncryptionByDefault(ctx, &ec2.GetEbsEncryptionByDefaultInput{}, testAccountID) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err) + + _, err = svc.EnableEbsEncryptionByDefault(ctx, &ec2.EnableEbsEncryptionByDefaultInput{}, testAccountID) + assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInternalError), "got %v", err) + + entry, err := svc.settingsKV.Get(ctx, testAccountID) + require.NoError(t, err) + var rec AccountSettingsRecord + require.NoError(t, json.Unmarshal(entry.Value(), &rec)) + assert.EqualValues(t, "quarantined", rec.State, "the unknown-state record was never overwritten") +} diff --git a/spinifex/handlers/iam/account_status_reader.go b/spinifex/handlers/iam/account_status_reader.go new file mode 100644 index 000000000..96e5d25b0 --- /dev/null +++ b/spinifex/handlers/iam/account_status_reader.go @@ -0,0 +1,56 @@ +package handlers_iam + +import ( + "context" + "encoding/json" + "errors" + + "github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore" + "github.com/nats-io/nats.go/jetstream" +) + +// AccountStatusReader reads account status from spinifex-accounts and never +// writes. It is the read-only capability EC2's settings seal uses to check +// its TERMINATING precondition. +type AccountStatusReader struct { + accounts *kvstore.Bucket // AttachOnly: never creates the bucket +} + +// NewAccountStatusReader attaches to the existing spinifex-accounts bucket. +// It needs no master key and runs no migration: a caller that only reads +// account status must not bootstrap IAM's bucket out from under it. +func NewAccountStatusReader(js jetstream.JetStream) *AccountStatusReader { + return &AccountStatusReader{ + accounts: kvstore.NewBucket(js, kvstore.Config{ + Name: KVBucketAccounts, + Missing: "account status reader: nil jetstream client", + AttachOnly: true, + }), + } +} + +// AccountTerminating is a direct read, which suffices because TERMINATING is +// one-way until the record is deleted. A missing record is (false, nil); only +// a failed read or decode is an error. +func (r *AccountStatusReader) AccountTerminating(ctx context.Context, accountID string) (bool, error) { + kv, err := r.accounts.KV(ctx) + if err != nil { + return false, err + } + + entry, err := kv.Get(ctx, accountID) + if err != nil { + if errors.Is(err, jetstream.ErrKeyNotFound) { + return false, nil + } + return false, err + } + + var record struct { + Status string `json:"status"` + } + if err := json.Unmarshal(entry.Value(), &record); err != nil { + return false, err + } + return record.Status == AccountStatusTerminating, nil +} diff --git a/spinifex/handlers/iam/account_status_reader_test.go b/spinifex/handlers/iam/account_status_reader_test.go new file mode 100644 index 000000000..08edb23b5 --- /dev/null +++ b/spinifex/handlers/iam/account_status_reader_test.go @@ -0,0 +1,102 @@ +package handlers_iam_test + +import ( + "testing" + + "github.com/mulgadc/spinifex/internal/testkit" + handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam" + "github.com/nats-io/nats.go/jetstream" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// newStatusReaderFixture starts a real IAM service (which creates and +// migrates spinifex-accounts) and a reader attached to the same JetStream. +func newStatusReaderFixture(t *testing.T) (handlers_iam.IAMService, *handlers_iam.AccountStatusReader, jetstream.JetStream) { + t.Helper() + _, nc, js := testutil.StartTestJetStream(t) + + masterKey, err := handlers_iam.GenerateMasterKey() + require.NoError(t, err) + svc, err := handlers_iam.NewIAMServiceImpl(t.Context(), nc, masterKey) + require.NoError(t, err) + + return svc, handlers_iam.NewAccountStatusReader(js), js +} + +func TestAccountStatusReader_TerminatingIsTrue(t *testing.T) { + t.Parallel() + svc, reader, _ := newStatusReaderFixture(t) + account, err := svc.CreateAccount("tenant@example.com") + require.NoError(t, err) + _, err = svc.SetAccountStatus(account.AccountID, handlers_iam.AccountStatusTerminating) + require.NoError(t, err) + + terminating, err := reader.AccountTerminating(t.Context(), account.AccountID) + require.NoError(t, err) + assert.True(t, terminating) +} + +func TestAccountStatusReader_ActiveAndSuspendedAreFalse(t *testing.T) { + t.Parallel() + svc, reader, _ := newStatusReaderFixture(t) + active, err := svc.CreateAccount("active@example.com") + require.NoError(t, err) + terminating, err := reader.AccountTerminating(t.Context(), active.AccountID) + require.NoError(t, err) + assert.False(t, terminating) + + suspended, err := svc.CreateAccount("suspended@example.com") + require.NoError(t, err) + _, err = svc.SetAccountStatus(suspended.AccountID, handlers_iam.AccountStatusSuspended) + require.NoError(t, err) + terminating, err = reader.AccountTerminating(t.Context(), suspended.AccountID) + require.NoError(t, err) + assert.False(t, terminating) +} + +func TestAccountStatusReader_MissingRecordIsFalseNotError(t *testing.T) { + t.Parallel() + _, reader, _ := newStatusReaderFixture(t) + terminating, err := reader.AccountTerminating(t.Context(), "999999999999") + require.NoError(t, err) + assert.False(t, terminating) +} + +// A missing bucket is an error, and the reader never creates one: a caller +// that only reads account status must not bootstrap IAM's bucket. +func TestAccountStatusReader_MissingBucketIsErrorAndNeverCreated(t *testing.T) { + t.Parallel() + _, nc, js := testutil.StartTestJetStream(t) + reader := handlers_iam.NewAccountStatusReader(js) + + _, err := reader.AccountTerminating(t.Context(), "111111111111") + require.Error(t, err) + + _, bucketErr := js.KeyValue(t.Context(), handlers_iam.KVBucketAccounts) + assert.ErrorIs(t, bucketErr, jetstream.ErrBucketNotFound) + _ = nc +} + +func TestAccountStatusReader_UndecodableValueIsError(t *testing.T) { + t.Parallel() + _, nc, js := testutil.StartTestJetStream(t) + _, err := handlers_iam.NewIAMServiceImpl(t.Context(), nc, mustMasterKey(t)) + require.NoError(t, err) + + kv, err := js.KeyValue(t.Context(), handlers_iam.KVBucketAccounts) + require.NoError(t, err) + _, err = kv.Put(t.Context(), "222222222222", []byte(`{"status":`)) + require.NoError(t, err) + + reader := handlers_iam.NewAccountStatusReader(js) + _, err = reader.AccountTerminating(t.Context(), "222222222222") + require.Error(t, err) +} + +func mustMasterKey(t *testing.T) []byte { + t.Helper() + key, err := handlers_iam.GenerateMasterKey() + require.NoError(t, err) + return key +}