mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
feat: account settings console access (#47)
This commit is contained in:
+3
-3
@@ -92,9 +92,9 @@ Persistence layer works (NATS JetStream KV), but stored values are not yet enfor
|
||||
| `enable-ebs-encryption-by-default` | — | `--dry-run` | None | Store encryption default flag in JetStream KV → return `EbsEncryptionByDefault=true` | 1. Enable encryption<br>2. Verify via get | **NOT STARTED** (enforcement pending) |
|
||||
| `disable-ebs-encryption-by-default` | — | `--dry-run` | None | Clear encryption default flag in JetStream KV → return `EbsEncryptionByDefault=false` | 1. Disable encryption<br>2. Verify via get | **NOT STARTED** (enforcement pending) |
|
||||
| `get-ebs-encryption-by-default` | — | `--dry-run` | None | Read encryption default flag from JetStream KV → return current state | 1. Get default state<br>2. Verify matches last enable/disable | **NOT STARTED** (enforcement pending) |
|
||||
| `enable-serial-console-access` | — | `--dry-run` | None | Store serial console access flag in JetStream KV → return enabled=true | 1. Enable access<br>2. Verify via get | **NOT STARTED** (enforcement pending) |
|
||||
| `disable-serial-console-access` | — | `--dry-run` | None | Clear serial console access flag in JetStream KV → return enabled=false | 1. Disable access | **NOT STARTED** (enforcement pending) |
|
||||
| `get-serial-console-access-status` | — | `--dry-run` | None | Read serial console access flag from JetStream KV | 1. Get current status | **NOT STARTED** (enforcement pending) |
|
||||
| `enable-serial-console-access` | — | `--dry-run` | None | Store serial console access flag in JetStream KV → return enabled=true | 1. Enable access<br>2. Verify via get | **DONE** |
|
||||
| `disable-serial-console-access` | — | `--dry-run` | None | Clear serial console access flag in JetStream KV → return enabled=false | 1. Disable access | **DONE** |
|
||||
| `get-serial-console-access-status` | — | `--dry-run` | None | Read serial console access flag from JetStream KV | 1. Get current status | **DONE** |
|
||||
| `enable-snapshot-block-public-access` | — | `--state` | None | Store snapshot block public access state in JetStream KV | 1. Enable with block-all-sharing<br>2. Enable with block-new-sharing | **NOT STARTED** (enforcement pending) |
|
||||
| `disable-snapshot-block-public-access` | — | `--dry-run` | None | Clear snapshot block public access state in JetStream KV | 1. Disable access | **NOT STARTED** (enforcement pending) |
|
||||
| `get-snapshot-block-public-access-state` | — | `--dry-run` | None | Read snapshot block public access state from JetStream KV | 1. Get current state | **NOT STARTED** (enforcement pending) |
|
||||
|
||||
@@ -367,6 +367,7 @@ var (
|
||||
ErrorSecurityGroupLimitExceeded = "SecurityGroupLimitExceeded"
|
||||
ErrorSecurityGroupsPerInstanceLimitExceeded = "SecurityGroupsPerInstanceLimitExceeded"
|
||||
ErrorSecurityGroupsPerInterfaceLimitExceeded = "SecurityGroupsPerInterfaceLimitExceeded"
|
||||
ErrorSerialConsoleSessionUnavailable = "SerialConsoleSessionUnavailable"
|
||||
ErrorServerInternal = "ServerInternal"
|
||||
ErrorServiceUnavailable = "ServiceUnavailable"
|
||||
ErrorSignatureDoesNotMatch = "SignatureDoesNotMatch"
|
||||
@@ -786,6 +787,7 @@ var ErrorLookup = map[string]ErrorMessage{
|
||||
ErrorSecurityGroupLimitExceeded: {HTTPCode: 400, Message: "You've reached the limit on the number of security groups that you can create, or that you can assign to an instance."},
|
||||
ErrorSecurityGroupsPerInstanceLimitExceeded: {HTTPCode: 400, Message: "You've reached the limit on the number of security groups that you can assign to an instance. For more information, see Amazon EC2 security groups."},
|
||||
ErrorSecurityGroupsPerInterfaceLimitExceeded: {HTTPCode: 400, Message: "You've reached the limit on the number of security groups you can associate with the specified network interface. For more information, see Amazon VPC quotas."},
|
||||
ErrorSerialConsoleSessionUnavailable: {HTTPCode: 403, Message: "The serial console access is not enabled for this account. Use EnableSerialConsoleAccess to enable access."},
|
||||
ErrorServerInternal: {HTTPCode: 500, Message: "An internal error has occurred. Retry your request, but if the problem persists, contact us with details by posting a message on AWS re:Post."},
|
||||
ErrorServiceUnavailable: {HTTPCode: 503, Message: "The request has failed due to a temporary failure of the server."},
|
||||
ErrorSignatureDoesNotMatch: {HTTPCode: 403, Message: "The request signature that Amazon has does not match the signature that you provided. Check your AWS credentials and signing method."},
|
||||
|
||||
@@ -369,6 +369,7 @@ func TestErrorLookup(t *testing.T) {
|
||||
{code: "SecurityGroupLimitExceeded", http: 400, message: "You've reached the limit on the number of security groups that you can create, or that you can assign to an instance."},
|
||||
{code: "SecurityGroupsPerInstanceLimitExceeded", http: 400, message: "You've reached the limit on the number of security groups that you can assign to an instance. For more information, see Amazon EC2 security groups."},
|
||||
{code: "SecurityGroupsPerInterfaceLimitExceeded", http: 400, message: "You've reached the limit on the number of security groups you can associate with the specified network interface. For more information, see Amazon VPC quotas."},
|
||||
{code: "SerialConsoleSessionUnavailable", http: 403, message: "The serial console access is not enabled for this account. Use EnableSerialConsoleAccess to enable access."},
|
||||
{code: "ServerInternal", http: 500, message: "An internal error has occurred. Retry your request, but if the problem persists, contact us with details by posting a message on AWS re:Post."},
|
||||
{code: "ServiceUnavailable", http: 503, message: "The request has failed due to a temporary failure of the server."},
|
||||
{code: "SignatureDoesNotMatch", http: 403, message: "The request signature that Amazon has does not match the signature that you provided. Check your AWS credentials and signing method."},
|
||||
|
||||
+27
-5
@@ -467,12 +467,9 @@ func (d *Daemon) Start() error {
|
||||
} else {
|
||||
d.eigwService = eigwSvc
|
||||
}
|
||||
accountSvc, err := handlers_ec2_account.NewAccountSettingsServiceImplWithNATS(d.config, d.natsConn)
|
||||
if err != nil {
|
||||
slog.Warn("Failed to create account settings service with NATS, using in-memory fallback", "error", err)
|
||||
accountSvc = handlers_ec2_account.NewAccountSettingsServiceImpl(d.config)
|
||||
if err := d.initAccountService(); err != nil {
|
||||
return fmt.Errorf("failed to initialize account settings service: %w", err)
|
||||
}
|
||||
d.accountService = accountSvc
|
||||
|
||||
// Protect daemon from OOM killer (prefer killing QEMU VMs instead)
|
||||
if err := utils.SetOOMScore(os.Getpid(), -500); err != nil {
|
||||
@@ -582,6 +579,31 @@ func (d *Daemon) initSnapshotService() (*handlers_ec2_snapshot.SnapshotServiceIm
|
||||
return nil, nil, fmt.Errorf("snapshot service unavailable after %d attempts", maxRetries)
|
||||
}
|
||||
|
||||
// initAccountService initializes the account settings service with retry/backoff.
|
||||
func (d *Daemon) initAccountService() error {
|
||||
const maxRetries = 10
|
||||
retryDelay := 500 * time.Millisecond
|
||||
|
||||
for attempt := 1; attempt <= maxRetries; attempt++ {
|
||||
svc, err := handlers_ec2_account.NewAccountSettingsServiceImplWithNATS(d.config, d.natsConn)
|
||||
if err == nil {
|
||||
if attempt > 1 {
|
||||
slog.Info("Account settings service initialized successfully", "attempts", attempt)
|
||||
}
|
||||
d.accountService = svc
|
||||
return nil
|
||||
}
|
||||
|
||||
slog.Warn("Failed to init account settings service", "error", err, "attempt", attempt, "maxRetries", maxRetries)
|
||||
if attempt < maxRetries {
|
||||
time.Sleep(retryDelay)
|
||||
retryDelay = min(retryDelay*2, 5*time.Second)
|
||||
}
|
||||
}
|
||||
|
||||
return fmt.Errorf("account settings service unavailable after %d attempts", maxRetries)
|
||||
}
|
||||
|
||||
// waitForClusterReady waits until dependent infrastructure services are reachable
|
||||
// before starting VM recovery. This prevents races where VMs try to mount volumes
|
||||
// before viperblock/predastore are ready.
|
||||
|
||||
@@ -1363,6 +1363,22 @@ func (d *Daemon) handleEC2GetConsoleOutput(msg *nats.Msg) {
|
||||
return
|
||||
}
|
||||
|
||||
// Check account-level serial console access
|
||||
status, err := d.accountService.GetSerialConsoleAccessStatus(&ec2.GetSerialConsoleAccessStatusInput{})
|
||||
if err != nil {
|
||||
slog.Error("Failed to check serial console access status", "err", err)
|
||||
if err := msg.Respond(utils.GenerateErrorPayload(awserrors.ErrorServerInternal)); err != nil {
|
||||
slog.Error("Failed to respond to NATS request", "err", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
if status.SerialConsoleAccessEnabled == nil || !*status.SerialConsoleAccessEnabled {
|
||||
if err := msg.Respond(utils.GenerateErrorPayload(awserrors.ErrorSerialConsoleSessionUnavailable)); err != nil {
|
||||
slog.Error("Failed to respond to NATS request", "err", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
instanceID := *input.InstanceId
|
||||
|
||||
// Find the instance on this node
|
||||
|
||||
@@ -26,6 +26,7 @@ import (
|
||||
"github.com/mulgadc/hive/hive/qmp"
|
||||
"github.com/mulgadc/hive/hive/vm"
|
||||
"github.com/mulgadc/viperblock/viperblock"
|
||||
"github.com/nats-io/nats-server/v2/server"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
@@ -45,7 +46,7 @@ func createFullTestDaemonWithStore(t *testing.T, natsURL string) (*Daemon, *obje
|
||||
daemon.snapshotService = handlers_ec2_snapshot.NewSnapshotServiceImplWithStore(cfg, memStore, daemon.natsConn)
|
||||
daemon.tagsService = handlers_ec2_tags.NewTagsServiceImplWithStore(cfg, memStore)
|
||||
daemon.eigwService = handlers_ec2_eigw.NewEgressOnlyIGWServiceImpl(cfg)
|
||||
daemon.accountService = handlers_ec2_account.NewAccountSettingsServiceImpl(cfg)
|
||||
initAccountServiceForTest(t, daemon)
|
||||
|
||||
return daemon, memStore
|
||||
}
|
||||
@@ -71,6 +72,32 @@ func createFullTestDaemonWithJetStream(t *testing.T, natsURL string) *Daemon {
|
||||
return daemon
|
||||
}
|
||||
|
||||
// initAccountServiceForTest initializes a JetStream-backed account service on the daemon
|
||||
// using an isolated embedded NATS JetStream server per test to avoid shared KV state.
|
||||
func initAccountServiceForTest(t *testing.T, daemon *Daemon) {
|
||||
t.Helper()
|
||||
ns, err := server.NewServer(&server.Options{
|
||||
Host: "127.0.0.1",
|
||||
Port: -1,
|
||||
JetStream: true,
|
||||
StoreDir: t.TempDir(),
|
||||
NoLog: true,
|
||||
NoSigs: true,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
go ns.Start()
|
||||
require.True(t, ns.ReadyForConnections(5*time.Second))
|
||||
t.Cleanup(func() { ns.Shutdown() })
|
||||
|
||||
nc, err := nats.Connect(ns.ClientURL())
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { nc.Close() })
|
||||
|
||||
svc, err := handlers_ec2_account.NewAccountSettingsServiceImplWithNATS(nil, nc)
|
||||
require.NoError(t, err)
|
||||
daemon.accountService = svc
|
||||
}
|
||||
|
||||
// --- handleNATSRequest generic tests ---
|
||||
|
||||
type testInput struct {
|
||||
@@ -1273,6 +1300,10 @@ func TestHandleEC2GetConsoleOutput(t *testing.T) {
|
||||
natsURL := sharedNATSURL
|
||||
daemon := createFullTestDaemon(t, natsURL)
|
||||
|
||||
// Enable serial console access
|
||||
_, err := daemon.accountService.EnableSerialConsoleAccess(&ec2.EnableSerialConsoleAccessInput{})
|
||||
require.NoError(t, err)
|
||||
|
||||
instanceID := "i-console-test-001"
|
||||
|
||||
// Create a temp console log file
|
||||
@@ -1321,6 +1352,10 @@ func TestHandleEC2GetConsoleOutput_EmptyLog(t *testing.T) {
|
||||
natsURL := sharedNATSURL
|
||||
daemon := createFullTestDaemon(t, natsURL)
|
||||
|
||||
// Enable serial console access
|
||||
_, err := daemon.accountService.EnableSerialConsoleAccess(&ec2.EnableSerialConsoleAccessInput{})
|
||||
require.NoError(t, err)
|
||||
|
||||
instanceID := "i-console-empty-001"
|
||||
|
||||
// Instance exists but no log file yet
|
||||
@@ -1358,6 +1393,10 @@ func TestHandleEC2GetConsoleOutput_NotFound(t *testing.T) {
|
||||
natsURL := sharedNATSURL
|
||||
daemon := createFullTestDaemon(t, natsURL)
|
||||
|
||||
// Enable serial console access
|
||||
_, err := daemon.accountService.EnableSerialConsoleAccess(&ec2.EnableSerialConsoleAccessInput{})
|
||||
require.NoError(t, err)
|
||||
|
||||
instanceID := "i-nonexistent-console"
|
||||
topic := fmt.Sprintf("ec2.%s.GetConsoleOutput", instanceID)
|
||||
sub, err := daemon.natsConn.Subscribe(topic, daemon.handleEC2GetConsoleOutput)
|
||||
@@ -1375,6 +1414,84 @@ func TestHandleEC2GetConsoleOutput_NotFound(t *testing.T) {
|
||||
assert.Contains(t, string(reply.Data), "InvalidInstanceID.NotFound")
|
||||
}
|
||||
|
||||
func TestHandleEC2GetConsoleOutput_SerialConsoleDisabled(t *testing.T) {
|
||||
natsURL := sharedNATSURL
|
||||
daemon := createFullTestDaemon(t, natsURL)
|
||||
|
||||
// Serial console access defaults to disabled — do NOT enable it
|
||||
|
||||
instanceID := "i-console-disabled-001"
|
||||
daemon.Instances.Mu.Lock()
|
||||
daemon.Instances.VMS[instanceID] = &vm.VM{
|
||||
ID: instanceID,
|
||||
Status: vm.StateRunning,
|
||||
Config: vm.Config{
|
||||
ConsoleLogPath: "/tmp/some-log.log",
|
||||
},
|
||||
}
|
||||
daemon.Instances.Mu.Unlock()
|
||||
|
||||
topic := fmt.Sprintf("ec2.%s.GetConsoleOutput", instanceID)
|
||||
sub, err := daemon.natsConn.Subscribe(topic, daemon.handleEC2GetConsoleOutput)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
input := &ec2.GetConsoleOutputInput{
|
||||
InstanceId: aws.String(instanceID),
|
||||
}
|
||||
reqData, _ := json.Marshal(input)
|
||||
reply, err := daemon.natsConn.Request(topic, reqData, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.Contains(t, string(reply.Data), "SerialConsoleSessionUnavailable")
|
||||
}
|
||||
|
||||
func TestHandleEC2GetConsoleOutput_EnableThenDisable(t *testing.T) {
|
||||
natsURL := sharedNATSURL
|
||||
daemon := createFullTestDaemon(t, natsURL)
|
||||
|
||||
instanceID := "i-console-toggle-001"
|
||||
tmpDir := t.TempDir()
|
||||
logPath := tmpDir + "/console.log"
|
||||
require.NoError(t, os.WriteFile(logPath, []byte("console output"), 0644))
|
||||
|
||||
daemon.Instances.Mu.Lock()
|
||||
daemon.Instances.VMS[instanceID] = &vm.VM{
|
||||
ID: instanceID,
|
||||
Status: vm.StateRunning,
|
||||
Config: vm.Config{
|
||||
ConsoleLogPath: logPath,
|
||||
},
|
||||
}
|
||||
daemon.Instances.Mu.Unlock()
|
||||
|
||||
topic := fmt.Sprintf("ec2.%s.GetConsoleOutput", instanceID)
|
||||
sub, err := daemon.natsConn.Subscribe(topic, daemon.handleEC2GetConsoleOutput)
|
||||
require.NoError(t, err)
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
// Enable serial console access — GetConsoleOutput should succeed
|
||||
_, err = daemon.accountService.EnableSerialConsoleAccess(&ec2.EnableSerialConsoleAccessInput{})
|
||||
require.NoError(t, err)
|
||||
|
||||
input := &ec2.GetConsoleOutputInput{InstanceId: aws.String(instanceID)}
|
||||
reqData, _ := json.Marshal(input)
|
||||
reply, err := daemon.natsConn.Request(topic, reqData, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
|
||||
var output ec2.GetConsoleOutputOutput
|
||||
require.NoError(t, json.Unmarshal(reply.Data, &output))
|
||||
assert.NotEmpty(t, *output.Output)
|
||||
|
||||
// Disable serial console access — GetConsoleOutput should fail
|
||||
_, err = daemon.accountService.DisableSerialConsoleAccess(&ec2.DisableSerialConsoleAccessInput{})
|
||||
require.NoError(t, err)
|
||||
|
||||
reply, err = daemon.natsConn.Request(topic, reqData, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, string(reply.Data), "SerialConsoleSessionUnavailable")
|
||||
}
|
||||
|
||||
// TestAttachVolume_ZoneMismatch verifies that attaching a volume in a different AZ
|
||||
// returns InvalidVolume.ZoneMismatch instead of proceeding.
|
||||
func TestAttachVolume_ZoneMismatch(t *testing.T) {
|
||||
|
||||
@@ -2,6 +2,7 @@ package handlers_ec2_account
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
|
||||
@@ -31,13 +32,6 @@ type AccountSettingsServiceImpl struct {
|
||||
settingsKV nats.KeyValue
|
||||
}
|
||||
|
||||
// NewAccountSettingsServiceImpl creates a new account settings service implementation
|
||||
func NewAccountSettingsServiceImpl(cfg *config.Config) *AccountSettingsServiceImpl {
|
||||
return &AccountSettingsServiceImpl{
|
||||
config: cfg,
|
||||
}
|
||||
}
|
||||
|
||||
// NewAccountSettingsServiceImplWithNATS creates an account settings service with NATS JetStream for persistence
|
||||
func NewAccountSettingsServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*AccountSettingsServiceImpl, error) {
|
||||
js, err := natsConn.JetStream()
|
||||
@@ -47,8 +41,7 @@ func NewAccountSettingsServiceImplWithNATS(cfg *config.Config, natsConn *nats.Co
|
||||
|
||||
settingsKV, err := getOrCreateKVBucket(js, KVBucketAccountSettings, 10)
|
||||
if err != nil {
|
||||
slog.Warn("Failed to create account settings KV bucket", "error", err)
|
||||
return NewAccountSettingsServiceImpl(cfg), nil
|
||||
return nil, fmt.Errorf("failed to create account settings KV bucket: %w", err)
|
||||
}
|
||||
|
||||
slog.Info("Account settings service initialized with JetStream KV", "bucket", KVBucketAccountSettings)
|
||||
@@ -76,21 +69,15 @@ func getOrCreateKVBucket(js nats.JetStreamContext, bucketName string, history in
|
||||
|
||||
// getSettings retrieves current account settings
|
||||
func (s *AccountSettingsServiceImpl) getSettings() (*AccountSettingsRecord, error) {
|
||||
if s.settingsKV == nil {
|
||||
// Return defaults if KV not available
|
||||
return &AccountSettingsRecord{
|
||||
EbsEncryptionByDefault: false,
|
||||
SerialConsoleAccess: false,
|
||||
}, nil
|
||||
}
|
||||
|
||||
entry, err := s.settingsKV.Get("default")
|
||||
if err != nil {
|
||||
// Return defaults if not found
|
||||
return &AccountSettingsRecord{
|
||||
EbsEncryptionByDefault: false,
|
||||
SerialConsoleAccess: false,
|
||||
}, nil
|
||||
if errors.Is(err, nats.ErrKeyNotFound) {
|
||||
return &AccountSettingsRecord{
|
||||
EbsEncryptionByDefault: false,
|
||||
SerialConsoleAccess: false,
|
||||
}, nil
|
||||
}
|
||||
return nil, fmt.Errorf("failed to get account settings: %w", err)
|
||||
}
|
||||
|
||||
var settings AccountSettingsRecord
|
||||
@@ -103,11 +90,6 @@ func (s *AccountSettingsServiceImpl) getSettings() (*AccountSettingsRecord, erro
|
||||
|
||||
// saveSettings saves current account settings
|
||||
func (s *AccountSettingsServiceImpl) saveSettings(settings *AccountSettingsRecord) error {
|
||||
if s.settingsKV == nil {
|
||||
slog.Warn("Cannot save settings - KV not initialized")
|
||||
return nil // Return success anyway for in-memory fallback
|
||||
}
|
||||
|
||||
data, err := json.Marshal(settings)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
@@ -11,6 +11,30 @@ import (
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// TestNewAccountSettingsServiceImplWithNATS_NoJetStream verifies that creating
|
||||
// the service against a NATS server without JetStream returns an error instead
|
||||
// of silently degrading.
|
||||
func TestNewAccountSettingsServiceImplWithNATS_NoJetStream(t *testing.T) {
|
||||
ns, err := server.NewServer(&server.Options{
|
||||
Host: "127.0.0.1",
|
||||
Port: -1,
|
||||
NoLog: true,
|
||||
NoSigs: true,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
go ns.Start()
|
||||
require.True(t, ns.ReadyForConnections(5*time.Second))
|
||||
t.Cleanup(func() { ns.Shutdown() })
|
||||
|
||||
nc, err := nats.Connect(ns.ClientURL())
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { nc.Close() })
|
||||
|
||||
svc, err := NewAccountSettingsServiceImplWithNATS(nil, nc)
|
||||
assert.Error(t, err)
|
||||
assert.Nil(t, svc)
|
||||
}
|
||||
|
||||
var _ AccountSettingsService = (*AccountSettingsServiceImpl)(nil)
|
||||
|
||||
func setupTestAccountService(t *testing.T) *AccountSettingsServiceImpl {
|
||||
|
||||
Reference in New Issue
Block a user