mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-23 20:57:02 +00:00
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.
This commit is contained in:
@@ -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.
|
||||
|
||||
@@ -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"`
|
||||
}
|
||||
@@ -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))
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user