diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index 630278ded..87911dc56 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -454,40 +454,55 @@ func (d *Daemon) Start() error { d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(d.config, d.resourceMgr.instanceTypes, d.natsConn, &d.Instances, store) d.keyService = handlers_ec2_key.NewKeyServiceImpl(d.config) d.imageService = handlers_ec2_image.NewImageServiceImpl(d.config, d.natsConn) - snapSvc, snapshotKV, err := d.initSnapshotService() - if err != nil { - return fmt.Errorf("failed to initialize snapshot service with NATS KV: %w", err) + + type snapResult struct { + svc *handlers_ec2_snapshot.SnapshotServiceImpl + kv nats.KeyValue } - d.snapshotService = snapSvc - d.volumeService = handlers_ec2_volume.NewVolumeServiceImpl(d.config, d.natsConn, snapshotKV) + snap, err := initServiceWithRetry("snapshot service", func() (snapResult, error) { + svc, kv, err := handlers_ec2_snapshot.NewSnapshotServiceImplWithNATS(d.config, d.natsConn) + return snapResult{svc, kv}, err + }) + if err != nil { + return fmt.Errorf("failed to initialize snapshot service: %w", err) + } + d.snapshotService = snap.svc + + d.volumeService = handlers_ec2_volume.NewVolumeServiceImpl(d.config, d.natsConn, snap.kv) d.tagsService = handlers_ec2_tags.NewTagsServiceImpl(d.config) - if eigwSvc, err := handlers_ec2_eigw.NewEgressOnlyIGWServiceImplWithNATS(d.config, d.natsConn); err != nil { - slog.Warn("Failed to initialize EIGW service, falling back to in-memory", "error", err) - d.eigwService = handlers_ec2_eigw.NewEgressOnlyIGWServiceImpl(d.config) - } else { - d.eigwService = eigwSvc + d.eigwService, err = initServiceWithRetry("EIGW service", func() (*handlers_ec2_eigw.EgressOnlyIGWServiceImpl, error) { + return handlers_ec2_eigw.NewEgressOnlyIGWServiceImplWithNATS(d.config, d.natsConn) + }) + if err != nil { + return fmt.Errorf("failed to initialize EIGW service: %w", err) } - if igwSvc, err := handlers_ec2_igw.NewIGWServiceImplWithNATS(d.config, d.natsConn); err != nil { - slog.Warn("Failed to initialize IGW service, falling back to in-memory", "error", err) - d.igwService = handlers_ec2_igw.NewIGWServiceImpl(d.config) - } else { - d.igwService = igwSvc + + d.igwService, err = initServiceWithRetry("IGW service", func() (*handlers_ec2_igw.IGWServiceImpl, error) { + return handlers_ec2_igw.NewIGWServiceImplWithNATS(d.config, d.natsConn) + }) + if err != nil { + return fmt.Errorf("failed to initialize IGW service: %w", err) } - if vpcSvc, err := handlers_ec2_vpc.NewVPCServiceImplWithNATS(d.config, d.natsConn); err != nil { - slog.Warn("Failed to initialize VPC service, falling back to in-memory", "error", err) - d.vpcService = handlers_ec2_vpc.NewVPCServiceImpl(d.config) - } else { - d.vpcService = vpcSvc + + d.vpcService, err = initServiceWithRetry("VPC service", func() (*handlers_ec2_vpc.VPCServiceImpl, error) { + return handlers_ec2_vpc.NewVPCServiceImplWithNATS(d.config, d.natsConn) + }) + if err != nil { + return fmt.Errorf("failed to initialize VPC service: %w", err) } - if err := d.initAccountService(); err != nil { + + d.accountService, err = initServiceWithRetry("account settings service", func() (*handlers_ec2_account.AccountSettingsServiceImpl, error) { + return handlers_ec2_account.NewAccountSettingsServiceImplWithNATS(d.config, d.natsConn) + }) + if err != nil { return fmt.Errorf("failed to initialize account settings service: %w", err) } // Ensure default VPC exists (matches AWS: every account has a default VPC) if d.vpcService != nil { - if err := d.vpcService.EnsureDefaultVPC(handlers_ec2_vpc.GlobalAccountID); err != nil { - slog.Warn("Failed to ensure default VPC", "error", err) + if err := d.vpcService.EnsureDefaultVPC(utils.GlobalAccountID); err != nil { + slog.Error("Failed to ensure default VPC", "error", err) } } @@ -603,10 +618,11 @@ func (d *Daemon) initJetStream() error { return nil } -// initSnapshotService initializes the snapshot service with retry/backoff. -// During cluster restarts, JetStream KV may be temporarily unavailable while -// NATS routes re-establish and the cluster forms quorum. -func (d *Daemon) initSnapshotService() (*handlers_ec2_snapshot.SnapshotServiceImpl, nats.KeyValue, error) { +// initServiceWithRetry initializes a service using the provided init function, +// retrying with exponential backoff (500ms→10s) for up to 5 minutes. During +// cluster restarts, JetStream KV may be temporarily unavailable while NATS +// routes re-establish and the cluster forms quorum. +func initServiceWithRetry[T any](name string, initFn func() (T, error)) (T, error) { const maxWait = 5 * time.Minute retryDelay := 500 * time.Millisecond start := time.Now() @@ -614,49 +630,21 @@ func (d *Daemon) initSnapshotService() (*handlers_ec2_snapshot.SnapshotServiceIm for { attempt++ - snapSvc, snapshotKV, err := handlers_ec2_snapshot.NewSnapshotServiceImplWithNATS(d.config, d.natsConn) + result, err := initFn() if err == nil { if attempt > 1 { - slog.Info("Snapshot service initialized successfully", "attempts", attempt, "elapsed", time.Since(start).Round(time.Second)) + slog.Info(name+" initialized successfully", "attempts", attempt, "elapsed", time.Since(start).Round(time.Second)) } - return snapSvc, snapshotKV, nil + return result, nil } elapsed := time.Since(start) if elapsed >= maxWait { - return nil, nil, fmt.Errorf("snapshot service unavailable after %s (%d attempts): %w", elapsed.Round(time.Second), attempt, err) + var zero T + return zero, fmt.Errorf("%s unavailable after %s (%d attempts): %w", name, elapsed.Round(time.Second), attempt, err) } - slog.Warn("Failed to init snapshot service", "error", err, "attempt", attempt, "elapsed", elapsed.Round(time.Second)) - time.Sleep(retryDelay) - retryDelay = min(retryDelay*2, 10*time.Second) - } -} - -// initAccountService initializes the account settings service with retry/backoff. -func (d *Daemon) initAccountService() error { - const maxWait = 5 * time.Minute - retryDelay := 500 * time.Millisecond - start := time.Now() - attempt := 0 - - for { - 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, "elapsed", time.Since(start).Round(time.Second)) - } - d.accountService = svc - return nil - } - - elapsed := time.Since(start) - if elapsed >= maxWait { - return fmt.Errorf("account settings service unavailable after %s (%d attempts): %w", elapsed.Round(time.Second), attempt, err) - } - - slog.Warn("Failed to init account settings service", "error", err, "attempt", attempt, "elapsed", elapsed.Round(time.Second)) + slog.Warn("Failed to init "+name, "error", err, "attempt", attempt, "elapsed", elapsed.Round(time.Second)) time.Sleep(retryDelay) retryDelay = min(retryDelay*2, 10*time.Second) } @@ -1384,7 +1372,7 @@ func (d *Daemon) stopInstance(instances map[string]*vm.VM, deleteVolume bool) er NetworkInterfaceId: &instance.ENIId, }, instance.AccountID) if eniErr != nil { - slog.Warn("Failed to delete ENI on termination", "eni", instance.ENIId, "err", eniErr) + slog.Error("Failed to delete ENI on termination", "eni", instance.ENIId, "err", eniErr) } else { slog.Info("Deleted ENI on termination", "eni", instance.ENIId, "instanceId", instance.ID) } diff --git a/hive/daemon/daemon_handlers.go b/hive/daemon/daemon_handlers.go index 786222fd1..47acdf674 100644 --- a/hive/daemon/daemon_handlers.go +++ b/hive/daemon/daemon_handlers.go @@ -73,11 +73,8 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) { return } - // Verify the caller owns this instance (skip for pre-Phase4 instances with no AccountID) - callerAccountID := utils.AccountIDFromMsg(msg) - if callerAccountID != "" && instance.AccountID != "" && instance.AccountID != callerAccountID { - slog.Warn("Account does not own this instance", "instanceId", command.ID, "callerAccount", callerAccountID, "ownerAccount", instance.AccountID) - respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound) + // Verify the caller owns this instance + if !checkInstanceOwnership(msg, command.ID, instance.AccountID) { return } diff --git a/hive/daemon/daemon_handlers_console.go b/hive/daemon/daemon_handlers_console.go index 7471b0cba..86567b905 100644 --- a/hive/daemon/daemon_handlers_console.go +++ b/hive/daemon/daemon_handlers_console.go @@ -32,7 +32,6 @@ func (d *Daemon) handleEC2GetConsoleOutput(msg *nats.Msg) { } instanceID := *input.InstanceId - accountID := utils.AccountIDFromMsg(msg) // Find the instance on this node d.Instances.Mu.Lock() @@ -45,8 +44,7 @@ func (d *Daemon) handleEC2GetConsoleOutput(msg *nats.Msg) { } // Verify the caller owns this instance - if accountID != "" && instance.AccountID != "" && instance.AccountID != accountID { - respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound) + if !checkInstanceOwnership(msg, instanceID, instance.AccountID) { return } diff --git a/hive/daemon/daemon_handlers_image.go b/hive/daemon/daemon_handlers_image.go index 769625daa..5998a799c 100644 --- a/hive/daemon/daemon_handlers_image.go +++ b/hive/daemon/daemon_handlers_image.go @@ -66,9 +66,7 @@ func (d *Daemon) handleEC2CreateImage(msg *nats.Msg) { } // Verify the caller owns this instance - if accountID != "" && instance.AccountID != "" && instance.AccountID != accountID { - slog.Warn("CreateImage: account does not own instance", "instanceId", instanceID, "accountID", accountID, "ownerAccountID", instance.AccountID) - respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound) + if !checkInstanceOwnership(msg, instanceID, instance.AccountID) { return } diff --git a/hive/daemon/daemon_handlers_instance.go b/hive/daemon/daemon_handlers_instance.go index 927fb17da..eb2c94aed 100644 --- a/hive/daemon/daemon_handlers_instance.go +++ b/hive/daemon/daemon_handlers_instance.go @@ -74,19 +74,7 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { // System AMIs have non-account-ID owner aliases (e.g. "self", "hive", empty). amiOwner := amiMeta.ImageOwnerAlias if amiOwner != "" && amiOwner != accountID { - // Check if this is a system/pre-phase4 AMI (non-account-ID owner) - isSystem := len(amiOwner) != 12 - if !isSystem { - allDigits := true - for _, c := range amiOwner { - if c < '0' || c > '9' { - allDigits = false - break - } - } - isSystem = !allDigits - } - if !isSystem { + if utils.IsAccountID(amiOwner) { slog.Warn("handleEC2RunInstances AMI not owned by caller", "imageId", *runInstancesInput.ImageId, "amiOwner", amiOwner, "accountID", accountID) respondWithError(msg, awserrors.ErrorInvalidAMIIDNotFound) return @@ -519,8 +507,8 @@ func (d *Daemon) handleEC2DescribeInstances(msg *nats.Msg) { // Iterate through all instances on this node for _, instance := range d.Instances.VMS { // Skip instances not owned by the caller's account. - // Pre-Phase4 instances (empty AccountID) are visible to all accounts. - if accountID != "" && instance.AccountID != "" && instance.AccountID != accountID { + // Pre-Phase4 instances (empty AccountID) are only visible to root. + if !isInstanceVisible(accountID, instance.AccountID) { continue } @@ -654,9 +642,6 @@ type startStoppedInstanceRequest struct { // handleEC2StartStoppedInstance picks up a stopped instance from shared KV, // re-launches it on this daemon node, and removes it from shared KV. func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) { - // Extract account ID from NATS header for ownership verification - callerAccountID := utils.AccountIDFromMsg(msg) - var req startStoppedInstanceRequest if err := json.Unmarshal(msg.Data, &req); err != nil { slog.Error("handleEC2StartStoppedInstance: failed to unmarshal request", "err", err) @@ -696,9 +681,7 @@ func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) { } // Verify the caller owns this instance - if callerAccountID != "" && instance.AccountID != "" && instance.AccountID != callerAccountID { - slog.Warn("handleEC2StartStoppedInstance: account does not own instance", "instanceId", req.InstanceID, "callerAccount", callerAccountID, "ownerAccount", instance.AccountID) - respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound) + if !checkInstanceOwnership(msg, req.InstanceID, instance.AccountID) { return } @@ -769,9 +752,6 @@ type terminateStoppedInstanceRequest struct { // handleEC2TerminateStoppedInstance picks up a stopped instance from shared KV, // deletes its volumes, and removes it from shared KV. func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) { - // Extract account ID from NATS header for ownership verification - callerAccountID := utils.AccountIDFromMsg(msg) - var req terminateStoppedInstanceRequest if err := json.Unmarshal(msg.Data, &req); err != nil { slog.Error("handleEC2TerminateStoppedInstance: failed to unmarshal request", "err", err) @@ -811,9 +791,7 @@ func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) { } // Verify the caller owns this instance - if callerAccountID != "" && instance.AccountID != "" && instance.AccountID != callerAccountID { - slog.Warn("handleEC2TerminateStoppedInstance: account does not own instance", "instanceId", req.InstanceID, "callerAccount", callerAccountID, "ownerAccount", instance.AccountID) - respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound) + if !checkInstanceOwnership(msg, req.InstanceID, instance.AccountID) { return } @@ -910,8 +888,8 @@ func (d *Daemon) handleEC2DescribeStoppedInstances(msg *nats.Msg) { for _, instance := range instances { // Skip instances not owned by the caller's account. - // Pre-Phase4 instances (empty AccountID) are visible to all accounts. - if accountID != "" && instance.AccountID != "" && instance.AccountID != accountID { + // Pre-Phase4 instances (empty AccountID) are only visible to root. + if !isInstanceVisible(accountID, instance.AccountID) { continue } @@ -977,9 +955,6 @@ func (d *Daemon) handleEC2DescribeStoppedInstances(msg *nats.Msg) { // handleEC2ModifyInstanceAttribute modifies attributes of a stopped instance in shared KV. // All supported attributes (InstanceType, UserData) require the instance to be stopped. func (d *Daemon) handleEC2ModifyInstanceAttribute(msg *nats.Msg) { - // Extract account ID from NATS header for ownership verification - callerAccountID := utils.AccountIDFromMsg(msg) - var input ec2.ModifyInstanceAttributeInput if err := json.Unmarshal(msg.Data, &input); err != nil { slog.Error("handleEC2ModifyInstanceAttribute: failed to unmarshal request", "err", err) @@ -1020,9 +995,7 @@ func (d *Daemon) handleEC2ModifyInstanceAttribute(msg *nats.Msg) { } // Verify the caller owns this instance - if callerAccountID != "" && instance.AccountID != "" && instance.AccountID != callerAccountID { - slog.Warn("handleEC2ModifyInstanceAttribute: account does not own instance", "instanceId", instanceID, "callerAccount", callerAccountID, "ownerAccount", instance.AccountID) - respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound) + if !checkInstanceOwnership(msg, instanceID, instance.AccountID) { return } diff --git a/hive/daemon/daemon_handlers_test.go b/hive/daemon/daemon_handlers_test.go index c81c37de6..2395c5a51 100644 --- a/hive/daemon/daemon_handlers_test.go +++ b/hive/daemon/daemon_handlers_test.go @@ -58,7 +58,6 @@ func createFullTestDaemonWithStore(t *testing.T, natsURL string) (*Daemon, *obje daemon.volumeService = handlers_ec2_volume.NewVolumeServiceImplWithStore(cfg, memStore, daemon.natsConn) 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) initAccountServiceForTest(t, daemon) return daemon, memStore @@ -543,6 +542,7 @@ func TestHandleEC2Events_StopInstance(t *testing.T) { Status: vm.StateRunning, Instance: &ec2.Instance{}, QMPClient: &qmp.QMPClient{}, + AccountID: testAccountID, } sub, err := daemon.natsConn.Subscribe( @@ -558,7 +558,7 @@ func TestHandleEC2Events_StopInstance(t *testing.T) { } cmdData, _ := json.Marshal(cmd) - reply, err := daemon.natsConn.Request( + reply, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), cmdData, 5*time.Second, @@ -588,6 +588,7 @@ func TestHandleEC2Events_TerminateInstance(t *testing.T) { Status: vm.StateRunning, Instance: &ec2.Instance{}, QMPClient: &qmp.QMPClient{}, + AccountID: testAccountID, } sub, err := daemon.natsConn.Subscribe( @@ -603,7 +604,7 @@ func TestHandleEC2Events_TerminateInstance(t *testing.T) { } cmdData, _ := json.Marshal(cmd) - reply, err := daemon.natsConn.Request( + reply, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), cmdData, 5*time.Second, @@ -914,8 +915,9 @@ func TestHandleEC2CreateImage_InvalidState(t *testing.T) { // Add an instance in "pending" state (not running or stopped) daemon.Instances.Mu.Lock() daemon.Instances.VMS["i-pending123"] = &vm.VM{ - ID: "i-pending123", - Status: vm.StatePending, + ID: "i-pending123", + Status: vm.StatePending, + AccountID: testAccountID, Instance: &ec2.Instance{ InstanceId: aws.String("i-pending123"), ImageId: aws.String("ami-source"), @@ -938,7 +940,7 @@ func TestHandleEC2CreateImage_InvalidState(t *testing.T) { Name: aws.String("my-image"), } reqData, _ := json.Marshal(input) - reply, err := daemon.natsConn.Request("ec2.CreateImage", reqData, 5*time.Second) + reply, err := natsRequest(daemon.natsConn, "ec2.CreateImage", reqData, 5*time.Second) require.NoError(t, err) var errResp map[string]any @@ -955,8 +957,9 @@ func TestHandleEC2CreateImage_NoRootVolume(t *testing.T) { // Add instance with no block device mappings daemon.Instances.Mu.Lock() daemon.Instances.VMS["i-novol123"] = &vm.VM{ - ID: "i-novol123", - Status: vm.StateRunning, + ID: "i-novol123", + Status: vm.StateRunning, + AccountID: testAccountID, Instance: &ec2.Instance{ InstanceId: aws.String("i-novol123"), ImageId: aws.String("ami-source"), @@ -974,7 +977,7 @@ func TestHandleEC2CreateImage_NoRootVolume(t *testing.T) { Name: aws.String("my-image"), } reqData, _ := json.Marshal(input) - reply, err := daemon.natsConn.Request("ec2.CreateImage", reqData, 5*time.Second) + reply, err := natsRequest(daemon.natsConn, "ec2.CreateImage", reqData, 5*time.Second) require.NoError(t, err) var errResp map[string]any @@ -1067,6 +1070,7 @@ func TestHandleEC2StartStoppedInstance_NotStoppedState(t *testing.T) { ID: "i-running-shared", Status: vm.StateRunning, InstanceType: getTestInstanceType(), + AccountID: testAccountID, } err := daemon.jsManager.WriteStoppedInstance(runningVM.ID, runningVM) require.NoError(t, err) @@ -1076,7 +1080,7 @@ func TestHandleEC2StartStoppedInstance_NotStoppedState(t *testing.T) { defer sub.Unsubscribe() reqData, _ := json.Marshal(map[string]string{"instance_id": "i-running-shared"}) - reply, err := daemon.natsConn.Request("ec2.start", reqData, 5*time.Second) + reply, err := natsRequest(daemon.natsConn, "ec2.start", reqData, 5*time.Second) require.NoError(t, err) var errResp map[string]any @@ -1101,6 +1105,7 @@ func TestHandleEC2DescribeStoppedInstances_ReturnsStoppedInstances(t *testing.T) Status: vm.StateStopped, InstanceType: getTestInstanceType(), LastNode: "node-1", + AccountID: testAccountID, Reservation: &ec2.Reservation{ ReservationId: aws.String("r-test-001"), OwnerId: aws.String("123456789012"), @@ -1119,7 +1124,7 @@ func TestHandleEC2DescribeStoppedInstances_ReturnsStoppedInstances(t *testing.T) input := &ec2.DescribeInstancesInput{} reqData, _ := json.Marshal(input) - reply, err := daemon.natsConn.Request("ec2.DescribeStoppedInstances", reqData, 5*time.Second) + reply, err := natsRequest(daemon.natsConn, "ec2.DescribeStoppedInstances", reqData, 5*time.Second) require.NoError(t, err) var output ec2.DescribeInstancesOutput @@ -1151,9 +1156,10 @@ func TestHandleEC2DescribeStoppedInstances_WithFilter(t *testing.T) { // Write two stopped instances for _, id := range []string{"i-filter-001", "i-filter-002"} { v := &vm.VM{ - ID: id, - Status: vm.StateStopped, - LastNode: "node-1", + ID: id, + Status: vm.StateStopped, + LastNode: "node-1", + AccountID: testAccountID, Reservation: &ec2.Reservation{ ReservationId: aws.String("r-filter"), OwnerId: aws.String("123456789012"), @@ -1175,7 +1181,7 @@ func TestHandleEC2DescribeStoppedInstances_WithFilter(t *testing.T) { InstanceIds: []*string{aws.String("i-filter-001")}, } reqData, _ := json.Marshal(input) - reply, err := daemon.natsConn.Request("ec2.DescribeStoppedInstances", reqData, 5*time.Second) + reply, err := natsRequest(daemon.natsConn, "ec2.DescribeStoppedInstances", reqData, 5*time.Second) require.NoError(t, err) var output ec2.DescribeInstancesOutput @@ -1252,6 +1258,7 @@ func TestHandleEC2TerminateStoppedInstance_NotStoppedState(t *testing.T) { ID: "i-term-running", Status: vm.StateRunning, InstanceType: getTestInstanceType(), + AccountID: testAccountID, } err := daemon.jsManager.WriteStoppedInstance(runningVM.ID, runningVM) require.NoError(t, err) @@ -1261,7 +1268,7 @@ func TestHandleEC2TerminateStoppedInstance_NotStoppedState(t *testing.T) { defer sub.Unsubscribe() reqData, _ := json.Marshal(map[string]string{"instance_id": "i-term-running"}) - reply, err := daemon.natsConn.Request("ec2.terminate", reqData, 5*time.Second) + reply, err := natsRequest(daemon.natsConn, "ec2.terminate", reqData, 5*time.Second) require.NoError(t, err) var errResp map[string]any @@ -1284,6 +1291,7 @@ func TestHandleEC2TerminateStoppedInstance_Success(t *testing.T) { Status: vm.StateStopped, InstanceType: getTestInstanceType(), LastNode: "node-1", + AccountID: testAccountID, Reservation: &ec2.Reservation{ ReservationId: aws.String("r-term-001"), OwnerId: aws.String("123456789012"), @@ -1301,7 +1309,7 @@ func TestHandleEC2TerminateStoppedInstance_Success(t *testing.T) { defer sub.Unsubscribe() reqData, _ := json.Marshal(map[string]string{"instance_id": "i-term-stopped-001"}) - reply, err := daemon.natsConn.Request("ec2.terminate", reqData, 5*time.Second) + reply, err := natsRequest(daemon.natsConn, "ec2.terminate", reqData, 5*time.Second) require.NoError(t, err) var resp map[string]string @@ -1330,8 +1338,9 @@ func TestHandleEC2GetConsoleOutput(t *testing.T) { // Add an instance with console log path daemon.Instances.Mu.Lock() daemon.Instances.VMS[instanceID] = &vm.VM{ - ID: instanceID, - Status: vm.StateRunning, + ID: instanceID, + Status: vm.StateRunning, + AccountID: testAccountID, Config: vm.Config{ ConsoleLogPath: logPath, }, @@ -1347,7 +1356,7 @@ func TestHandleEC2GetConsoleOutput(t *testing.T) { InstanceId: aws.String(instanceID), } reqData, _ := json.Marshal(input) - reply, err := daemon.natsConn.Request(topic, reqData, 5*time.Second) + reply, err := natsRequest(daemon.natsConn, topic, reqData, 5*time.Second) require.NoError(t, err) var output ec2.GetConsoleOutputOutput @@ -1373,8 +1382,9 @@ func TestHandleEC2GetConsoleOutput_EmptyLog(t *testing.T) { // Instance exists but no log file yet daemon.Instances.Mu.Lock() daemon.Instances.VMS[instanceID] = &vm.VM{ - ID: instanceID, - Status: vm.StateRunning, + ID: instanceID, + Status: vm.StateRunning, + AccountID: testAccountID, Config: vm.Config{ ConsoleLogPath: "/nonexistent/console.log", }, @@ -1390,7 +1400,7 @@ func TestHandleEC2GetConsoleOutput_EmptyLog(t *testing.T) { InstanceId: aws.String(instanceID), } reqData, _ := json.Marshal(input) - reply, err := daemon.natsConn.Request(topic, reqData, 5*time.Second) + reply, err := natsRequest(daemon.natsConn, topic, reqData, 5*time.Second) require.NoError(t, err) var output ec2.GetConsoleOutputOutput @@ -1439,6 +1449,7 @@ func TestAttachVolume_ZoneMismatch(t *testing.T) { ID: instanceID, InstanceType: getTestInstanceType(), Status: vm.StateRunning, + AccountID: testAccountID, Instance: &ec2.Instance{}, QMPClient: &qmp.QMPClient{}, } @@ -1483,7 +1494,7 @@ func TestAttachVolume_ZoneMismatch(t *testing.T) { } cmdData, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), cmdData, 5*time.Second, @@ -1508,6 +1519,7 @@ func TestHandleEC2ModifyInstanceAttribute_ChangeInstanceType(t *testing.T) { ID: instanceID, Status: vm.StateStopped, InstanceType: "t3.micro", + AccountID: testAccountID, Config: vm.Config{InstanceType: "t3.micro"}, Instance: &ec2.Instance{ InstanceId: aws.String(instanceID), @@ -1523,7 +1535,7 @@ func TestHandleEC2ModifyInstanceAttribute_ChangeInstanceType(t *testing.T) { InstanceType: &ec2.AttributeValue{Value: aws.String("t3.medium")}, } reqData, _ := json.Marshal(input) - reply, err := daemon.natsConn.Request("ec2.ModifyInstanceAttribute", reqData, 5*time.Second) + reply, err := natsRequest(daemon.natsConn, "ec2.ModifyInstanceAttribute", reqData, 5*time.Second) require.NoError(t, err) assert.Equal(t, `{}`, string(reply.Data)) @@ -1549,6 +1561,7 @@ func TestHandleEC2ModifyInstanceAttribute_ChangeUserData(t *testing.T) { ID: instanceID, Status: vm.StateStopped, InstanceType: "t3.micro", + AccountID: testAccountID, UserData: "old data", RunInstancesInput: &ec2.RunInstancesInput{ UserData: aws.String("b2xkIGRhdGE="), @@ -1569,7 +1582,7 @@ func TestHandleEC2ModifyInstanceAttribute_ChangeUserData(t *testing.T) { UserData: &ec2.BlobAttributeValue{Value: []byte(newContent)}, } reqData, _ := json.Marshal(input) - reply, err := daemon.natsConn.Request("ec2.ModifyInstanceAttribute", reqData, 5*time.Second) + reply, err := natsRequest(daemon.natsConn, "ec2.ModifyInstanceAttribute", reqData, 5*time.Second) require.NoError(t, err) assert.Equal(t, `{}`, string(reply.Data)) @@ -1617,6 +1630,7 @@ func TestHandleEC2ModifyInstanceAttribute_NotStopped(t *testing.T) { ID: instanceID, Status: vm.StateRunning, InstanceType: "t3.micro", + AccountID: testAccountID, } err = daemon.jsManager.WriteStoppedInstance(instanceID, instance) require.NoError(t, err) @@ -1627,7 +1641,7 @@ func TestHandleEC2ModifyInstanceAttribute_NotStopped(t *testing.T) { InstanceType: &ec2.AttributeValue{Value: aws.String("t3.medium")}, } reqData, _ := json.Marshal(input) - reply, err := daemon.natsConn.Request("ec2.ModifyInstanceAttribute", reqData, 5*time.Second) + reply, err := natsRequest(daemon.natsConn, "ec2.ModifyInstanceAttribute", reqData, 5*time.Second) require.NoError(t, err) var errResp map[string]any @@ -1650,6 +1664,7 @@ func TestHandleEC2ModifyInstanceAttribute_ClearsStateReason(t *testing.T) { ID: instanceID, Status: vm.StateStopped, InstanceType: "m7i.small", + AccountID: testAccountID, Config: vm.Config{InstanceType: "m7i.small"}, Instance: &ec2.Instance{ InstanceId: aws.String(instanceID), @@ -1669,7 +1684,7 @@ func TestHandleEC2ModifyInstanceAttribute_ClearsStateReason(t *testing.T) { InstanceType: &ec2.AttributeValue{Value: aws.String("t3.micro")}, } reqData, _ := json.Marshal(input) - reply, err := daemon.natsConn.Request("ec2.ModifyInstanceAttribute", reqData, 5*time.Second) + reply, err := natsRequest(daemon.natsConn, "ec2.ModifyInstanceAttribute", reqData, 5*time.Second) require.NoError(t, err) assert.Equal(t, `{}`, string(reply.Data)) @@ -1696,6 +1711,7 @@ func TestHandleEC2ModifyInstanceAttribute_InvalidTypeAccepted(t *testing.T) { ID: instanceID, Status: vm.StateStopped, InstanceType: "t3.micro", + AccountID: testAccountID, Config: vm.Config{InstanceType: "t3.micro"}, Instance: &ec2.Instance{ InstanceId: aws.String(instanceID), @@ -1712,7 +1728,7 @@ func TestHandleEC2ModifyInstanceAttribute_InvalidTypeAccepted(t *testing.T) { InstanceType: &ec2.AttributeValue{Value: aws.String("z99.mega")}, } reqData, _ := json.Marshal(input) - reply, err := daemon.natsConn.Request("ec2.ModifyInstanceAttribute", reqData, 5*time.Second) + reply, err := natsRequest(daemon.natsConn, "ec2.ModifyInstanceAttribute", reqData, 5*time.Second) require.NoError(t, err) assert.Equal(t, `{}`, string(reply.Data)) @@ -2518,6 +2534,7 @@ func TestHandleEC2TerminateStoppedInstance_WithVolumes(t *testing.T) { stoppedVM := &vm.VM{ ID: "i-term-vol-001", Status: vm.StateStopped, + AccountID: testAccountID, InstanceType: getTestInstanceType(), LastNode: "node-1", Reservation: &ec2.Reservation{ @@ -2545,7 +2562,7 @@ func TestHandleEC2TerminateStoppedInstance_WithVolumes(t *testing.T) { defer sub.Unsubscribe() reqData, _ := json.Marshal(map[string]string{"instance_id": "i-term-vol-001"}) - reply, err := daemon.natsConn.Request("ec2.terminate", reqData, 10*time.Second) + reply, err := natsRequest(daemon.natsConn, "ec2.terminate", reqData, 10*time.Second) require.NoError(t, err) var resp map[string]string @@ -2616,6 +2633,7 @@ func TestHandleEC2StartStoppedInstance_InstanceTypeNotAvailable(t *testing.T) { ID: "i-start-badtype-001", Status: vm.StateStopped, InstanceType: "z99.nonexistent", + AccountID: testAccountID, Instance: &ec2.Instance{ InstanceId: aws.String("i-start-badtype-001"), InstanceType: aws.String("z99.nonexistent"), @@ -2630,7 +2648,7 @@ func TestHandleEC2StartStoppedInstance_InstanceTypeNotAvailable(t *testing.T) { defer sub.Unsubscribe() reqData, _ := json.Marshal(map[string]string{"instance_id": "i-start-badtype-001"}) - reply, err := daemon.natsConn.Request("ec2.start", reqData, 5*time.Second) + reply, err := natsRequest(daemon.natsConn, "ec2.start", reqData, 5*time.Second) require.NoError(t, err) var errResp map[string]any @@ -2672,6 +2690,7 @@ func TestHandleEC2CreateImage_RunningInstanceReachesService(t *testing.T) { ID: instanceID, Status: vm.StateRunning, InstanceType: getTestInstanceType(), + AccountID: testAccountID, Instance: &ec2.Instance{ InstanceId: aws.String(instanceID), ImageId: aws.String(sourceImageID), @@ -2699,7 +2718,7 @@ func TestHandleEC2CreateImage_RunningInstanceReachesService(t *testing.T) { Name: aws.String("test-image-snapshot"), } reqData, _ := json.Marshal(input) - reply, err := daemon.natsConn.Request( + reply, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.%s.CreateImage", instanceID), reqData, 5*time.Second, @@ -2720,6 +2739,7 @@ func TestAttachVolume_MissingVolumeData(t *testing.T) { instance := &vm.VM{ ID: instanceID, Status: vm.StateRunning, + AccountID: testAccountID, InstanceType: getTestInstanceType(), Instance: &ec2.Instance{}, QMPClient: &qmp.QMPClient{}, @@ -2743,7 +2763,7 @@ func TestAttachVolume_MissingVolumeData(t *testing.T) { } cmdData, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), cmdData, 5*time.Second, @@ -2759,6 +2779,7 @@ func TestAttachVolume_InstanceNotRunning(t *testing.T) { instance := &vm.VM{ ID: instanceID, Status: vm.StateStopped, + AccountID: testAccountID, InstanceType: getTestInstanceType(), Instance: &ec2.Instance{}, QMPClient: &qmp.QMPClient{}, @@ -2783,7 +2804,7 @@ func TestAttachVolume_InstanceNotRunning(t *testing.T) { } cmdData, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), cmdData, 5*time.Second, @@ -2799,6 +2820,7 @@ func TestAttachVolume_VolumeNotFound(t *testing.T) { instance := &vm.VM{ ID: instanceID, Status: vm.StateRunning, + AccountID: testAccountID, InstanceType: getTestInstanceType(), Instance: &ec2.Instance{}, QMPClient: &qmp.QMPClient{}, @@ -2823,7 +2845,7 @@ func TestAttachVolume_VolumeNotFound(t *testing.T) { } cmdData, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), cmdData, 5*time.Second, @@ -2841,6 +2863,7 @@ func TestAttachVolume_VolumeInUse(t *testing.T) { instance := &vm.VM{ ID: instanceID, Status: vm.StateRunning, + AccountID: testAccountID, InstanceType: getTestInstanceType(), Instance: &ec2.Instance{}, QMPClient: &qmp.QMPClient{}, @@ -2884,7 +2907,7 @@ func TestAttachVolume_VolumeInUse(t *testing.T) { } cmdData, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), cmdData, 5*time.Second, @@ -2902,6 +2925,7 @@ func TestDetachVolume_MissingVolumeData(t *testing.T) { instance := &vm.VM{ ID: instanceID, Status: vm.StateRunning, + AccountID: testAccountID, InstanceType: getTestInstanceType(), Instance: &ec2.Instance{}, QMPClient: &qmp.QMPClient{}, @@ -2924,7 +2948,7 @@ func TestDetachVolume_MissingVolumeData(t *testing.T) { } cmdData, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), cmdData, 5*time.Second, @@ -2940,6 +2964,7 @@ func TestDetachVolume_InstanceNotRunning(t *testing.T) { instance := &vm.VM{ ID: instanceID, Status: vm.StateStopped, + AccountID: testAccountID, InstanceType: getTestInstanceType(), Instance: &ec2.Instance{}, QMPClient: &qmp.QMPClient{}, @@ -2964,7 +2989,7 @@ func TestDetachVolume_InstanceNotRunning(t *testing.T) { } cmdData, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), cmdData, 5*time.Second, @@ -2980,6 +3005,7 @@ func TestDetachVolume_VolumeNotAttached(t *testing.T) { instance := &vm.VM{ ID: instanceID, Status: vm.StateRunning, + AccountID: testAccountID, InstanceType: getTestInstanceType(), Instance: &ec2.Instance{}, QMPClient: &qmp.QMPClient{}, @@ -3004,7 +3030,7 @@ func TestDetachVolume_VolumeNotAttached(t *testing.T) { } cmdData, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), cmdData, 5*time.Second, @@ -3020,6 +3046,7 @@ func TestDetachVolume_BootVolumeRejected(t *testing.T) { instance := &vm.VM{ ID: instanceID, Status: vm.StateRunning, + AccountID: testAccountID, InstanceType: getTestInstanceType(), Instance: &ec2.Instance{}, QMPClient: &qmp.QMPClient{}, @@ -3047,7 +3074,7 @@ func TestDetachVolume_BootVolumeRejected(t *testing.T) { } cmdData, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), cmdData, 5*time.Second, @@ -3063,6 +3090,7 @@ func TestDetachVolume_DeviceMismatch(t *testing.T) { instance := &vm.VM{ ID: instanceID, Status: vm.StateRunning, + AccountID: testAccountID, InstanceType: getTestInstanceType(), Instance: &ec2.Instance{}, QMPClient: &qmp.QMPClient{}, @@ -3091,7 +3119,7 @@ func TestDetachVolume_DeviceMismatch(t *testing.T) { } cmdData, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), cmdData, 5*time.Second, diff --git a/hive/daemon/daemon_test.go b/hive/daemon/daemon_test.go index 4c6f0f390..be6c12af7 100644 --- a/hive/daemon/daemon_test.go +++ b/hive/daemon/daemon_test.go @@ -860,12 +860,14 @@ func TestDaemon_BootAllocation(t *testing.T) { ID: "i-running", InstanceType: getTestInstanceType(), Status: vm.StateRunning, + AccountID: testAccountID, Attributes: qmp.Attributes{StopInstance: false}, }, "i-stopped": { ID: "i-stopped", InstanceType: getTestInstanceType(), Status: vm.StateStopped, + AccountID: testAccountID, Attributes: qmp.Attributes{StopInstance: true}, }, "i-terminated": { @@ -940,6 +942,7 @@ func TestStopInstance_Deallocation(t *testing.T) { ID: instanceId, InstanceType: instanceTypeStr, Status: vm.StateRunning, + AccountID: testAccountID, } err := daemon.resourceMgr.allocate(instanceType) @@ -1123,6 +1126,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) { daemon.Instances.VMS[instanceID] = &vm.VM{ ID: instanceID, Status: vm.StateRunning, + AccountID: testAccountID, Reservation: reservation1, Instance: ec2Instance, } @@ -1142,6 +1146,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) { daemon.Instances.VMS[instanceID] = &vm.VM{ ID: instanceID, Status: vm.StateRunning, + AccountID: testAccountID, Reservation: reservation2, Instance: ec2Instance, } @@ -1159,6 +1164,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) { daemon.Instances.VMS["i-single-001"] = &vm.VM{ ID: "i-single-001", Status: vm.StateStopped, + AccountID: testAccountID, Reservation: reservation3, Instance: ec2Instance, } @@ -1172,7 +1178,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) { input := &ec2.DescribeInstancesInput{} inputJSON, _ := json.Marshal(input) - resp, err := daemon.natsConn.Request("ec2.DescribeInstances", inputJSON, 5*time.Second) + resp, err := natsRequest(daemon.natsConn, "ec2.DescribeInstances", inputJSON, 5*time.Second) require.NoError(t, err) var output ec2.DescribeInstancesOutput @@ -1202,7 +1208,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) { } inputJSON, _ := json.Marshal(input) - resp, err := daemon.natsConn.Request("ec2.DescribeInstances", inputJSON, 5*time.Second) + resp, err := natsRequest(daemon.natsConn, "ec2.DescribeInstances", inputJSON, 5*time.Second) require.NoError(t, err) var output ec2.DescribeInstancesOutput @@ -1226,7 +1232,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) { } inputJSON, _ := json.Marshal(input) - resp, err := daemon.natsConn.Request("ec2.DescribeInstances", inputJSON, 5*time.Second) + resp, err := natsRequest(daemon.natsConn, "ec2.DescribeInstances", inputJSON, 5*time.Second) require.NoError(t, err) var output ec2.DescribeInstancesOutput @@ -1250,7 +1256,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) { } inputJSON, _ := json.Marshal(input) - resp, err := daemon.natsConn.Request("ec2.DescribeInstances", inputJSON, 5*time.Second) + resp, err := natsRequest(daemon.natsConn, "ec2.DescribeInstances", inputJSON, 5*time.Second) require.NoError(t, err) var output ec2.DescribeInstancesOutput @@ -1268,7 +1274,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) { input := &ec2.DescribeInstancesInput{} inputJSON, _ := json.Marshal(input) - resp, err := daemon.natsConn.Request("ec2.DescribeInstances", inputJSON, 5*time.Second) + resp, err := natsRequest(daemon.natsConn, "ec2.DescribeInstances", inputJSON, 5*time.Second) require.NoError(t, err) var output ec2.DescribeInstancesOutput @@ -1752,6 +1758,7 @@ func TestStopInstance_DeleteOnTermination_VolumeDeletion(t *testing.T) { ID: "i-test-dot", InstanceType: getTestInstanceType(), Status: vm.StateRunning, + AccountID: testAccountID, QMPClient: &qmp.QMPClient{}, // nil encoder/decoder => QMP will fail, which is fine EBSRequests: config.EBSRequests{ Requests: []config.EBSRequest{ @@ -1849,6 +1856,7 @@ func TestStopInstance_DeleteOnTermination_False_SkipsVolumeDeletion(t *testing.T ID: "i-test-no-delete", InstanceType: getTestInstanceType(), Status: vm.StateRunning, + AccountID: testAccountID, QMPClient: &qmp.QMPClient{}, EBSRequests: config.EBSRequests{ Requests: []config.EBSRequest{ @@ -1935,6 +1943,7 @@ func TestStopInstance_NoDelete_OnStop(t *testing.T) { ID: "i-test-stop-only", InstanceType: getTestInstanceType(), Status: vm.StateRunning, + AccountID: testAccountID, QMPClient: &qmp.QMPClient{}, EBSRequests: config.EBSRequests{ Requests: []config.EBSRequest{ @@ -1986,6 +1995,7 @@ func TestHandleEC2Events_AttachVolume(t *testing.T) { ID: instanceID, InstanceType: instanceType, Status: vm.StateRunning, + AccountID: testAccountID, Instance: &ec2.Instance{}, QMPClient: &qmp.QMPClient{}, // nil encoder/decoder } @@ -2009,7 +2019,7 @@ func TestHandleEC2Events_AttachVolume(t *testing.T) { } data, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), data, 5*time.Second, @@ -2037,7 +2047,7 @@ func TestHandleEC2Events_AttachVolume(t *testing.T) { } data, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), data, 5*time.Second, @@ -2065,7 +2075,7 @@ func TestHandleEC2Events_AttachVolume(t *testing.T) { } data, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), data, 5*time.Second, @@ -2114,6 +2124,7 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) { ID: instanceID, InstanceType: instanceType, Status: vm.StateRunning, + AccountID: testAccountID, Instance: &ec2.Instance{ BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{ { @@ -2154,7 +2165,7 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) { } data, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), data, 5*time.Second, @@ -2180,7 +2191,7 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) { } data, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), data, 5*time.Second, @@ -2206,7 +2217,7 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) { } data, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), data, 5*time.Second, @@ -2237,7 +2248,7 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) { } data, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), data, 5*time.Second, @@ -2272,7 +2283,7 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) { } data, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), data, 5*time.Second, @@ -2306,7 +2317,7 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) { } data, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), data, 5*time.Second, @@ -2332,7 +2343,7 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) { } data, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), data, 5*time.Second, @@ -2356,7 +2367,7 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) { } data, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), data, 5*time.Second, @@ -2450,6 +2461,7 @@ func TestDetachVolume_SuccessPath(t *testing.T) { ID: instanceID, InstanceType: instanceType, Status: vm.StateRunning, + AccountID: testAccountID, Instance: &ec2.Instance{ BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{ { @@ -2515,7 +2527,7 @@ func TestDetachVolume_SuccessPath(t *testing.T) { } data, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), data, 10*time.Second, @@ -2603,6 +2615,7 @@ func TestDetachVolume_ForceFlag(t *testing.T) { ID: instanceID, InstanceType: instanceType, Status: vm.StateRunning, + AccountID: testAccountID, Instance: &ec2.Instance{ BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{ { @@ -2654,7 +2667,7 @@ func TestDetachVolume_ForceFlag(t *testing.T) { } data, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), data, 10*time.Second, @@ -2716,6 +2729,7 @@ func TestDetachVolume_BlockdevDelFailure(t *testing.T) { ID: instanceID, InstanceType: instanceType, Status: vm.StateRunning, + AccountID: testAccountID, Instance: &ec2.Instance{ BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{ { @@ -2757,7 +2771,7 @@ func TestDetachVolume_BlockdevDelFailure(t *testing.T) { } data, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), data, 10*time.Second, @@ -2807,6 +2821,7 @@ func TestDetachVolume_SuccessWithDeviceMatch(t *testing.T) { ID: instanceID, InstanceType: instanceType, Status: vm.StateRunning, + AccountID: testAccountID, Instance: &ec2.Instance{ BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{ { @@ -2857,7 +2872,7 @@ func TestDetachVolume_SuccessWithDeviceMatch(t *testing.T) { } data, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), data, 10*time.Second, @@ -2891,6 +2906,7 @@ func TestAttachVolume_ReplacesStaleEBSRequest(t *testing.T) { ID: instanceID, InstanceType: instanceType, Status: vm.StateRunning, + AccountID: testAccountID, Instance: &ec2.Instance{}, QMPClient: qmpClient, EBSRequests: config.EBSRequests{ @@ -2933,7 +2949,7 @@ func TestAttachVolume_ReplacesStaleEBSRequest(t *testing.T) { } data, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), data, 10*time.Second, @@ -3236,6 +3252,7 @@ func TestMarkInstanceFailed(t *testing.T) { ID: instanceID, InstanceType: getTestInstanceType(), Status: vm.StatePending, + AccountID: testAccountID, Instance: ec2Instance, QMPClient: &qmp.QMPClient{}, } @@ -3267,6 +3284,7 @@ func TestMarkInstanceFailed_NilInstance(t *testing.T) { ID: instanceID, InstanceType: getTestInstanceType(), Status: vm.StatePending, + AccountID: testAccountID, Instance: nil, // no ec2.Instance QMPClient: &qmp.QMPClient{}, } @@ -3389,7 +3407,7 @@ func TestDescribeInstances_InvalidInstanceIDMalformed(t *testing.T) { } inputJSON, _ := json.Marshal(input) - resp, err := daemon.natsConn.Request("ec2.DescribeInstances", inputJSON, 5*time.Second) + resp, err := natsRequest(daemon.natsConn, "ec2.DescribeInstances", inputJSON, 5*time.Second) require.NoError(t, err) assert.Contains(t, string(resp.Data), "InvalidInstanceID.Malformed") }) @@ -3400,7 +3418,7 @@ func TestDescribeInstances_InvalidInstanceIDMalformed(t *testing.T) { } inputJSON, _ := json.Marshal(input) - resp, err := daemon.natsConn.Request("ec2.DescribeInstances", inputJSON, 5*time.Second) + resp, err := natsRequest(daemon.natsConn, "ec2.DescribeInstances", inputJSON, 5*time.Second) require.NoError(t, err) // Should not contain a malformed error — returns empty results instead assert.NotContains(t, string(resp.Data), "InvalidInstanceID.Malformed") @@ -3419,6 +3437,7 @@ func TestStopTerminate_IncorrectInstanceState(t *testing.T) { ID: instanceID, InstanceType: getTestInstanceType(), Status: vm.StateStopped, + AccountID: testAccountID, Instance: &ec2.Instance{}, } daemon.Instances.VMS[instanceID] = instance @@ -3443,7 +3462,7 @@ func TestStopTerminate_IncorrectInstanceState(t *testing.T) { } data, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), data, 5*time.Second, @@ -3466,7 +3485,7 @@ func TestStopTerminate_IncorrectInstanceState(t *testing.T) { } data, _ := json.Marshal(command) - resp, err := daemon.natsConn.Request( + resp, err := natsRequest(daemon.natsConn, fmt.Sprintf("ec2.cmd.%s", instanceID), data, 5*time.Second, diff --git a/hive/daemon/ownership.go b/hive/daemon/ownership.go new file mode 100644 index 000000000..910f821fc --- /dev/null +++ b/hive/daemon/ownership.go @@ -0,0 +1,45 @@ +package daemon + +import ( + "log/slog" + + "github.com/mulgadc/hive/hive/awserrors" + "github.com/mulgadc/hive/hive/utils" + "github.com/nats-io/nats.go" +) + +// checkInstanceOwnership verifies the caller owns the instance. +// Returns true if access is allowed, false if denied (error already sent). +// Pre-Phase4 instances (empty AccountID) are only visible to root (GlobalAccountID). +func checkInstanceOwnership(msg *nats.Msg, instanceID, ownerAccountID string) bool { + callerAccountID := utils.AccountIDFromMsg(msg) + + // Pre-Phase4 instance: only root can access + if ownerAccountID == "" { + if callerAccountID != utils.GlobalAccountID { + slog.Warn("Pre-Phase4 instance access denied (not root)", + "instanceId", instanceID, "callerAccount", callerAccountID) + respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound) + return false + } + return true + } + + // Normal ownership check + if callerAccountID != ownerAccountID { + slog.Warn("Account does not own instance", + "instanceId", instanceID, "callerAccount", callerAccountID, "ownerAccount", ownerAccountID) + respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound) + return false + } + return true +} + +// isInstanceVisible checks if the caller can see this instance (for Describe operations). +// Pre-Phase4 instances (empty AccountID) are only visible to root (GlobalAccountID). +func isInstanceVisible(callerAccountID, ownerAccountID string) bool { + if ownerAccountID == "" { + return callerAccountID == utils.GlobalAccountID + } + return callerAccountID == ownerAccountID +} diff --git a/hive/gateway/auth_test.go b/hive/gateway/auth_test.go index d6350e0bb..86c81ead0 100644 --- a/hive/gateway/auth_test.go +++ b/hive/gateway/auth_test.go @@ -13,6 +13,7 @@ import ( "github.com/gofiber/fiber/v2" "github.com/mulgadc/hive/hive/awserrors" handlers_iam "github.com/mulgadc/hive/hive/handlers/iam" + "github.com/mulgadc/hive/hive/utils" "github.com/mulgadc/predastore/auth" ) @@ -1461,7 +1462,7 @@ func TestCheckPolicy_RootGlobalAccount_Bypasses(t *testing.T) { AccessKeyID: testAccessKey, SecretAccessKey: encryptedSecret, UserName: "root", - AccountID: handlers_iam.GlobalAccountID, + AccountID: utils.GlobalAccountID, Status: "Active", }, }, diff --git a/hive/gateway/ec2.go b/hive/gateway/ec2.go index 166d4c235..dd9780e10 100644 --- a/hive/gateway/ec2.go +++ b/hive/gateway/ec2.go @@ -98,7 +98,7 @@ var ec2Actions = map[string]EC2Handler{ return gateway_ec2_image.DescribeImages(input, gw.NATSConn, accountID) }), "CreateImage": ec2Handler(func(input *ec2.CreateImageInput, gw *GatewayConfig, accountID string) (any, error) { - return gateway_ec2_image.CreateImage(input, gw.NATSConn, accountID) + return gateway_ec2_image.CreateImage(input, gw.NATSConn, gw.DiscoverActiveNodes(), accountID) }), "DescribeRegions": ec2Handler(func(input *ec2.DescribeRegionsInput, gw *GatewayConfig, accountID string) (any, error) { return gateway_ec2_zone.DescribeRegions(input, gw.Region) diff --git a/hive/gateway/ec2/image/CreateImage.go b/hive/gateway/ec2/image/CreateImage.go index 0f524dc1b..f515378ff 100644 --- a/hive/gateway/ec2/image/CreateImage.go +++ b/hive/gateway/ec2/image/CreateImage.go @@ -32,14 +32,14 @@ func ValidateCreateImageInput(input *ec2.CreateImageInput) error { } // CreateImage handles the EC2 CreateImage API call -func CreateImage(input *ec2.CreateImageInput, natsConn *nats.Conn, accountID string) (ec2.CreateImageOutput, error) { +func CreateImage(input *ec2.CreateImageInput, natsConn *nats.Conn, expectedNodes int, accountID string) (ec2.CreateImageOutput, error) { var output ec2.CreateImageOutput if err := ValidateCreateImageInput(input); err != nil { return output, err } - svc := handlers_ec2_image.NewNATSImageService(natsConn) + svc := handlers_ec2_image.NewNATSImageService(natsConn, expectedNodes) result, err := svc.CreateImage(input, accountID) if err != nil { return output, err diff --git a/hive/gateway/ec2/image/DescribeImages.go b/hive/gateway/ec2/image/DescribeImages.go index e3c46c79e..4c4183fac 100644 --- a/hive/gateway/ec2/image/DescribeImages.go +++ b/hive/gateway/ec2/image/DescribeImages.go @@ -37,7 +37,7 @@ func DescribeImages(input *ec2.DescribeImagesInput, natsConn *nats.Conn, account } // Create NATS service and call handler - imageService := handlers_ec2_image.NewNATSImageService(natsConn) + imageService := handlers_ec2_image.NewNATSImageService(natsConn, 0) result, err := imageService.DescribeImages(input, accountID) if err != nil { diff --git a/hive/gateway/ec2/instance/RunInstances.go b/hive/gateway/ec2/instance/RunInstances.go index fe6e9ff51..781dc5324 100644 --- a/hive/gateway/ec2/instance/RunInstances.go +++ b/hive/gateway/ec2/instance/RunInstances.go @@ -90,7 +90,7 @@ func RunInstances(input *ec2.RunInstancesInput, natsConn *nats.Conn, accountID s // isKnownInstanceType checks whether any daemon recognizes the given instance type. func isKnownInstanceType(natsConn *nats.Conn, instanceType string) bool { result, err := utils.NATSRequest[ec2.DescribeInstanceTypesOutput]( - natsConn, "ec2.DescribeInstanceTypes", &ec2.DescribeInstanceTypesInput{}, 3*time.Second, "") + natsConn, "ec2.DescribeInstanceTypes", &ec2.DescribeInstanceTypesInput{}, 3*time.Second, utils.GlobalAccountID) if err != nil || result == nil { return false } diff --git a/hive/gateway/ec2/volume/AttachVolume.go b/hive/gateway/ec2/volume/AttachVolume.go index 672851437..77f0b1811 100644 --- a/hive/gateway/ec2/volume/AttachVolume.go +++ b/hive/gateway/ec2/volume/AttachVolume.go @@ -74,7 +74,7 @@ func AttachVolume(input *ec2.AttachVolumeInput, natsConn *nats.Conn, accountID s slog.Error("AttachVolume: NATS request failed", "instanceId", instanceID, "volumeId", volumeID, "err", err) if errors.Is(err, nats.ErrNoResponders) { // Check shared KV for stopped instances before returning NotFound - if isStoppedInstance(instanceID, natsConn) { + if isStoppedInstance(instanceID, natsConn, accountID) { return output, errors.New(awserrors.ErrorIncorrectInstanceState) } return output, errors.New(awserrors.ErrorInvalidInstanceIDNotFound) @@ -98,7 +98,7 @@ func AttachVolume(input *ec2.AttachVolumeInput, natsConn *nats.Conn, accountID s // isStoppedInstance checks the shared KV (via ec2.DescribeStoppedInstances) to // determine whether instanceID exists as a stopped instance. -func isStoppedInstance(instanceID string, natsConn *nats.Conn) bool { +func isStoppedInstance(instanceID string, natsConn *nats.Conn, accountID string) bool { input := &ec2.DescribeInstancesInput{ InstanceIds: []*string{aws.String(instanceID)}, } @@ -107,7 +107,10 @@ func isStoppedInstance(instanceID string, natsConn *nats.Conn) bool { return false } - msg, err := natsConn.Request("ec2.DescribeStoppedInstances", reqData, 3*time.Second) + reqMsg := nats.NewMsg("ec2.DescribeStoppedInstances") + reqMsg.Data = reqData + reqMsg.Header.Set(utils.AccountIDHeader, accountID) + msg, err := natsConn.RequestMsg(reqMsg, 3*time.Second) if err != nil { return false } diff --git a/hive/gateway/gateway.go b/hive/gateway/gateway.go index 88a849f12..85459b6c5 100644 --- a/hive/gateway/gateway.go +++ b/hive/gateway/gateway.go @@ -19,6 +19,7 @@ import ( "github.com/mulgadc/hive/hive/awserrors" "github.com/mulgadc/hive/hive/gateway/policy" handlers_iam "github.com/mulgadc/hive/hive/handlers/iam" + "github.com/mulgadc/hive/hive/utils" "github.com/nats-io/nats.go" ) @@ -199,7 +200,7 @@ func (gw *GatewayConfig) checkPolicy(ctx *fiber.Ctx, service, action string) err return errors.New(awserrors.ErrorInternalError) } - if identity == "" || (identity == "root" && accountID == handlers_iam.GlobalAccountID) { + if identity == "" || (identity == "root" && accountID == utils.GlobalAccountID) { return nil } diff --git a/hive/handlers/ec2/account/service_impl.go b/hive/handlers/ec2/account/service_impl.go index 9cfca51f4..0b45d18c1 100644 --- a/hive/handlers/ec2/account/service_impl.go +++ b/hive/handlers/ec2/account/service_impl.go @@ -39,7 +39,7 @@ func NewAccountSettingsServiceImplWithNATS(cfg *config.Config, natsConn *nats.Co return nil, fmt.Errorf("failed to get JetStream context: %w", err) } - settingsKV, err := getOrCreateKVBucket(js, KVBucketAccountSettings, 10) + settingsKV, err := utils.GetOrCreateKVBucket(js, KVBucketAccountSettings, 10) if err != nil { return nil, fmt.Errorf("failed to create account settings KV bucket: %w", err) } @@ -53,25 +53,11 @@ func NewAccountSettingsServiceImplWithNATS(cfg *config.Config, natsConn *nats.Co }, nil } -func getOrCreateKVBucket(js nats.JetStreamContext, bucketName string, history int) (nats.KeyValue, error) { - kv, err := js.CreateKeyValue(&nats.KeyValueConfig{ - Bucket: bucketName, - History: utils.SafeIntToUint8(history), - }) - if err != nil { - kv, err = js.KeyValue(bucketName) - if err != nil { - return nil, err - } - } - return kv, nil -} - // settingsKey returns the per-account KV key for storing settings. -// Falls back to "default" for pre-Phase-4 resources with no accountID. +// Falls back to GlobalAccountID for pre-Phase-4 resources with no accountID. func settingsKey(accountID string) string { if accountID == "" { - return "default" + return utils.GlobalAccountID } return accountID } diff --git a/hive/handlers/ec2/account/service_impl_test.go b/hive/handlers/ec2/account/service_impl_test.go index c73b1b90e..1e4253bc5 100644 --- a/hive/handlers/ec2/account/service_impl_test.go +++ b/hive/handlers/ec2/account/service_impl_test.go @@ -5,6 +5,7 @@ import ( "time" "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/hive/hive/utils" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" @@ -169,6 +170,6 @@ func TestSerialConsole_MultiAccountIsolation(t *testing.T) { } func TestSettingsKey_EmptyAccountIDFallback(t *testing.T) { - assert.Equal(t, "default", settingsKey("")) + assert.Equal(t, utils.GlobalAccountID, settingsKey("")) assert.Equal(t, "123456789012", settingsKey("123456789012")) } diff --git a/hive/handlers/ec2/eigw/service_impl.go b/hive/handlers/ec2/eigw/service_impl.go index c5601f1f2..d0a2c58d4 100644 --- a/hive/handlers/ec2/eigw/service_impl.go +++ b/hive/handlers/ec2/eigw/service_impl.go @@ -12,6 +12,7 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/hive/hive/awserrors" "github.com/mulgadc/hive/hive/config" + handlers_ec2_vpc "github.com/mulgadc/hive/hive/handlers/ec2/vpc" "github.com/mulgadc/hive/hive/utils" "github.com/nats-io/nats.go" ) @@ -20,7 +21,6 @@ import ( var _ EgressOnlyIGWService = (*EgressOnlyIGWServiceImpl)(nil) const KVBucketEgressOnlyIGW = "hive-eigw" -const kvBucketVPCs = "hive-vpc-vpcs" // EgressOnlyIGWRecord represents a stored Egress-only Internet Gateway type EgressOnlyIGWRecord struct { @@ -38,13 +38,6 @@ type EgressOnlyIGWServiceImpl struct { vpcKV nats.KeyValue } -// NewEgressOnlyIGWServiceImpl creates a new Egress-only Internet Gateway service implementation -func NewEgressOnlyIGWServiceImpl(cfg *config.Config) *EgressOnlyIGWServiceImpl { - return &EgressOnlyIGWServiceImpl{ - config: cfg, - } -} - // NewEgressOnlyIGWServiceImplWithNATS creates an Egress-only Internet Gateway service with NATS JetStream for persistence func NewEgressOnlyIGWServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*EgressOnlyIGWServiceImpl, error) { js, err := natsConn.JetStream() @@ -52,15 +45,15 @@ func NewEgressOnlyIGWServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn return nil, fmt.Errorf("failed to get JetStream context: %w", err) } - eigwKV, err := getOrCreateKVBucket(js, KVBucketEgressOnlyIGW, 10) + eigwKV, err := utils.GetOrCreateKVBucket(js, KVBucketEgressOnlyIGW, 10) if err != nil { return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketEgressOnlyIGW, err) } - // Get VPC KV bucket for cross-resource ownership validation - vpcKV, err := js.KeyValue(kvBucketVPCs) + // Get or create VPC KV bucket for cross-resource ownership validation + vpcKV, err := utils.GetOrCreateKVBucket(js, handlers_ec2_vpc.KVBucketVPCs, 10) if err != nil { - slog.Warn("EIGW service: VPC KV bucket not available, VPC ownership checks disabled", "error", err) + return nil, fmt.Errorf("failed to get VPC KV bucket: %w", err) } slog.Info("Egress-only IGW service initialized with JetStream KV", "bucket", KVBucketEgressOnlyIGW) @@ -72,36 +65,20 @@ func NewEgressOnlyIGWServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn }, nil } -func getOrCreateKVBucket(js nats.JetStreamContext, bucketName string, history int) (nats.KeyValue, error) { - kv, err := js.CreateKeyValue(&nats.KeyValueConfig{ - Bucket: bucketName, - History: utils.SafeIntToUint8(history), - }) - if err != nil { - kv, err = js.KeyValue(bucketName) - if err != nil { - return nil, err - } - } - return kv, nil -} - -func accountKey(accountID, resourceID string) string { - return accountID + "." + resourceID -} - // CreateEgressOnlyInternetGateway creates a new Egress-only Internet Gateway func (s *EgressOnlyIGWServiceImpl) CreateEgressOnlyInternetGateway(input *ec2.CreateEgressOnlyInternetGatewayInput, accountID string) (*ec2.CreateEgressOnlyInternetGatewayOutput, error) { if input.VpcId == nil || *input.VpcId == "" { return nil, errors.New(awserrors.ErrorMissingParameter) } - // Verify the caller owns the target VPC - if s.vpcKV != nil { - if _, err := s.vpcKV.Get(accountKey(accountID, *input.VpcId)); err != nil { - slog.Warn("CreateEgressOnlyInternetGateway: VPC not found for account", "vpcId", *input.VpcId, "accountID", accountID) - return nil, errors.New(awserrors.ErrorInvalidVpcIDNotFound) - } + // Verify the caller owns the target VPC (fail-closed if KV unavailable) + if s.vpcKV == nil { + slog.Error("VPC KV unavailable, cannot verify VPC ownership") + return nil, errors.New(awserrors.ErrorServerInternal) + } + if _, err := s.vpcKV.Get(utils.AccountKey(accountID, *input.VpcId)); err != nil { + slog.Warn("CreateEgressOnlyInternetGateway: VPC not found for account", "vpcId", *input.VpcId, "accountID", accountID) + return nil, errors.New(awserrors.ErrorInvalidVpcIDNotFound) } eigwID := utils.GenerateResourceID("eigw") @@ -128,7 +105,7 @@ func (s *EgressOnlyIGWServiceImpl) CreateEgressOnlyInternetGateway(input *ec2.Cr if err != nil { return nil, fmt.Errorf("failed to marshal Egress-only IGW record: %w", err) } - if _, err := s.eigwKV.Put(accountKey(accountID, eigwID), data); err != nil { + if _, err := s.eigwKV.Put(utils.AccountKey(accountID, eigwID), data); err != nil { return nil, errors.New(awserrors.ErrorServerInternal) } @@ -146,12 +123,15 @@ func (s *EgressOnlyIGWServiceImpl) DeleteEgressOnlyInternetGateway(input *ec2.De } eigwID := *input.EgressOnlyInternetGatewayId - key := accountKey(accountID, eigwID) + key := utils.AccountKey(accountID, eigwID) + // Verify the EIGW exists before deleting if _, err := s.eigwKV.Get(key); err != nil { - return nil, errors.New(awserrors.ErrorInvalidEgressOnlyInternetGatewayIdNotFound) + if errors.Is(err, nats.ErrKeyNotFound) { + return nil, errors.New(awserrors.ErrorInvalidEgressOnlyInternetGatewayIdNotFound) + } + return nil, errors.New(awserrors.ErrorServerInternal) } - if err := s.eigwKV.Delete(key); err != nil { return nil, errors.New(awserrors.ErrorServerInternal) } diff --git a/hive/handlers/ec2/eigw/service_impl_test.go b/hive/handlers/ec2/eigw/service_impl_test.go index df3d7030b..91a419d6f 100644 --- a/hive/handlers/ec2/eigw/service_impl_test.go +++ b/hive/handlers/ec2/eigw/service_impl_test.go @@ -6,6 +6,8 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" + handlers_ec2_vpc "github.com/mulgadc/hive/hive/handlers/ec2/vpc" + "github.com/mulgadc/hive/hive/utils" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" @@ -34,6 +36,16 @@ func setupTestEIGWService(t *testing.T) *EgressOnlyIGWServiceImpl { require.NoError(t, err) t.Cleanup(func() { nc.Close() }) + // Create VPC KV bucket and register test VPCs so fail-closed ownership checks pass + js, err := nc.JetStream() + require.NoError(t, err) + vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1}) + require.NoError(t, err) + for _, vpcID := range []string{"vpc-test123", "vpc-tagged", "vpc-tagged2"} { + _, err = vpcKV.Put(utils.AccountKey(testAccountID, vpcID), []byte(`{"vpc_id":"`+vpcID+`","state":"available"}`)) + require.NoError(t, err) + } + svc, err := NewEgressOnlyIGWServiceImplWithNATS(nil, nc) require.NoError(t, err) return svc @@ -211,11 +223,11 @@ func TestCreateEgressOnlyInternetGateway_CrossAccountVPCRejected(t *testing.T) { // Create VPC KV bucket with a VPC owned by testAccountID js, err := nc.JetStream() require.NoError(t, err) - vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: kvBucketVPCs, History: 1}) + vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1}) require.NoError(t, err) vpcID := "vpc-alpha456" - _, err = vpcKV.Put(accountKey(testAccountID, vpcID), []byte(`{"vpc_id":"vpc-alpha456","state":"available"}`)) + _, err = vpcKV.Put(utils.AccountKey(testAccountID, vpcID), []byte(`{"vpc_id":"vpc-alpha456","state":"available"}`)) require.NoError(t, err) // Create EIGW service (VPC KV will be populated since we created it before EIGW init) diff --git a/hive/handlers/ec2/igw/service_impl.go b/hive/handlers/ec2/igw/service_impl.go index ae09c3b6a..39f6926f9 100644 --- a/hive/handlers/ec2/igw/service_impl.go +++ b/hive/handlers/ec2/igw/service_impl.go @@ -12,6 +12,7 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/hive/hive/awserrors" "github.com/mulgadc/hive/hive/config" + handlers_ec2_vpc "github.com/mulgadc/hive/hive/handlers/ec2/vpc" "github.com/mulgadc/hive/hive/utils" "github.com/nats-io/nats.go" ) @@ -20,7 +21,6 @@ import ( var _ IGWService = (*IGWServiceImpl)(nil) const KVBucketIGW = "hive-igw" -const kvBucketVPCs = "hive-vpc-vpcs" // IGWRecord represents a stored Internet Gateway type IGWRecord struct { @@ -39,13 +39,6 @@ type IGWServiceImpl struct { natsConn *nats.Conn } -// NewIGWServiceImpl creates a new Internet Gateway service without NATS persistence -func NewIGWServiceImpl(cfg *config.Config) *IGWServiceImpl { - return &IGWServiceImpl{ - config: cfg, - } -} - // NewIGWServiceImplWithNATS creates an Internet Gateway service with NATS JetStream for persistence func NewIGWServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*IGWServiceImpl, error) { js, err := natsConn.JetStream() @@ -53,15 +46,15 @@ func NewIGWServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*IGWSer return nil, fmt.Errorf("failed to get JetStream context: %w", err) } - igwKV, err := getOrCreateKVBucket(js, KVBucketIGW, 10) + igwKV, err := utils.GetOrCreateKVBucket(js, KVBucketIGW, 10) if err != nil { return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketIGW, err) } - // Get VPC KV bucket for cross-resource ownership validation - vpcKV, err := js.KeyValue(kvBucketVPCs) + // Get or create VPC KV bucket for cross-resource ownership validation + vpcKV, err := utils.GetOrCreateKVBucket(js, handlers_ec2_vpc.KVBucketVPCs, 10) if err != nil { - slog.Warn("IGW service: VPC KV bucket not available, VPC ownership checks disabled", "error", err) + return nil, fmt.Errorf("failed to get VPC KV bucket: %w", err) } slog.Info("IGW service initialized with JetStream KV", "bucket", KVBucketIGW) @@ -74,24 +67,6 @@ func NewIGWServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*IGWSer }, nil } -func getOrCreateKVBucket(js nats.JetStreamContext, bucketName string, history int) (nats.KeyValue, error) { - kv, err := js.CreateKeyValue(&nats.KeyValueConfig{ - Bucket: bucketName, - History: utils.SafeIntToUint8(history), - }) - if err != nil { - kv, err = js.KeyValue(bucketName) - if err != nil { - return nil, err - } - } - return kv, nil -} - -func accountKey(accountID, resourceID string) string { - return accountID + "." + resourceID -} - // CreateInternetGateway creates a new Internet Gateway (initially detached) func (s *IGWServiceImpl) CreateInternetGateway(input *ec2.CreateInternetGatewayInput, accountID string) (*ec2.CreateInternetGatewayOutput, error) { igwID := utils.GenerateResourceID("igw") @@ -117,7 +92,7 @@ func (s *IGWServiceImpl) CreateInternetGateway(input *ec2.CreateInternetGatewayI if err != nil { return nil, fmt.Errorf("failed to marshal IGW record: %w", err) } - if _, err := s.igwKV.Put(accountKey(accountID, igwID), data); err != nil { + if _, err := s.igwKV.Put(utils.AccountKey(accountID, igwID), data); err != nil { return nil, errors.New(awserrors.ErrorServerInternal) } @@ -135,7 +110,7 @@ func (s *IGWServiceImpl) DeleteInternetGateway(input *ec2.DeleteInternetGatewayI } igwID := *input.InternetGatewayId - key := accountKey(accountID, igwID) + key := utils.AccountKey(accountID, igwID) entry, err := s.igwKV.Get(key) if err != nil { @@ -231,7 +206,7 @@ func (s *IGWServiceImpl) AttachInternetGateway(input *ec2.AttachInternetGatewayI igwID := *input.InternetGatewayId vpcID := *input.VpcId - key := accountKey(accountID, igwID) + key := utils.AccountKey(accountID, igwID) entry, err := s.igwKV.Get(key) if err != nil { @@ -247,12 +222,14 @@ func (s *IGWServiceImpl) AttachInternetGateway(input *ec2.AttachInternetGatewayI return nil, errors.New(awserrors.ErrorResourceAlreadyAssociated) } - // Verify the caller owns the target VPC - if s.vpcKV != nil { - if _, err := s.vpcKV.Get(accountKey(accountID, vpcID)); err != nil { - slog.Warn("AttachInternetGateway: VPC not found for account", "vpcId", vpcID, "accountID", accountID) - return nil, errors.New(awserrors.ErrorInvalidVpcIDNotFound) - } + // Verify the caller owns the target VPC (fail-closed if KV unavailable) + if s.vpcKV == nil { + slog.Error("VPC KV unavailable, cannot verify VPC ownership") + return nil, errors.New(awserrors.ErrorServerInternal) + } + if _, err := s.vpcKV.Get(utils.AccountKey(accountID, vpcID)); err != nil { + slog.Warn("AttachInternetGateway: VPC not found for account", "vpcId", vpcID, "accountID", accountID) + return nil, errors.New(awserrors.ErrorInvalidVpcIDNotFound) } record.VpcId = vpcID @@ -295,7 +272,7 @@ func (s *IGWServiceImpl) DetachInternetGateway(input *ec2.DetachInternetGatewayI igwID := *input.InternetGatewayId vpcID := *input.VpcId - key := accountKey(accountID, igwID) + key := utils.AccountKey(accountID, igwID) entry, err := s.igwKV.Get(key) if err != nil { diff --git a/hive/handlers/ec2/igw/service_impl_test.go b/hive/handlers/ec2/igw/service_impl_test.go index e0af84716..0213cc592 100644 --- a/hive/handlers/ec2/igw/service_impl_test.go +++ b/hive/handlers/ec2/igw/service_impl_test.go @@ -6,6 +6,8 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" + handlers_ec2_vpc "github.com/mulgadc/hive/hive/handlers/ec2/vpc" + "github.com/mulgadc/hive/hive/utils" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" @@ -32,6 +34,16 @@ func setupTestIGWService(t *testing.T) (*IGWServiceImpl, *nats.Conn) { require.NoError(t, err) t.Cleanup(func() { nc.Close() }) + // Create VPC KV bucket and register test VPCs so fail-closed ownership checks pass + js, err := nc.JetStream() + require.NoError(t, err) + vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1}) + require.NoError(t, err) + for _, vpcID := range []string{"vpc-test123", "vpc-other", "vpc-lifecycle", "vpc-event-test"} { + _, err = vpcKV.Put(utils.AccountKey(testAccountID, vpcID), []byte(`{"vpc_id":"`+vpcID+`","state":"available"}`)) + require.NoError(t, err) + } + svc, err := NewIGWServiceImplWithNATS(nil, nc) require.NoError(t, err) return svc, nc @@ -443,14 +455,6 @@ func TestCreateInternetGateway_PublishesNoEvent(t *testing.T) { } } -func TestNewIGWServiceImpl(t *testing.T) { - svc := NewIGWServiceImpl(nil) - require.NotNil(t, svc) - // Without KV, IGW operations will panic (nil KV), so just verify constructor - assert.Nil(t, svc.igwKV) - assert.Nil(t, svc.natsConn) -} - // TestAttachInternetGateway_CrossAccountVPCRejected tests that attaching an IGW to another account's VPC is rejected. func TestAttachInternetGateway_CrossAccountVPCRejected(t *testing.T) { svc, nc := setupTestIGWService(t) @@ -458,11 +462,11 @@ func TestAttachInternetGateway_CrossAccountVPCRejected(t *testing.T) { // Create VPC KV bucket and add a VPC owned by testAccountID js, err := nc.JetStream() require.NoError(t, err) - vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: kvBucketVPCs, History: 1}) + vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1}) require.NoError(t, err) vpcID := "vpc-alpha123" - _, err = vpcKV.Put(accountKey(testAccountID, vpcID), []byte(`{"vpc_id":"vpc-alpha123","state":"available"}`)) + _, err = vpcKV.Put(utils.AccountKey(testAccountID, vpcID), []byte(`{"vpc_id":"vpc-alpha123","state":"available"}`)) require.NoError(t, err) // Refresh service to pick up the VPC KV bucket diff --git a/hive/handlers/ec2/igw/service_nats_test.go b/hive/handlers/ec2/igw/service_nats_test.go index 61692f298..d607e3666 100644 --- a/hive/handlers/ec2/igw/service_nats_test.go +++ b/hive/handlers/ec2/igw/service_nats_test.go @@ -6,6 +6,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/hive/hive/utils" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -44,7 +45,7 @@ func handleNATSMsg[In any, Out any](msg *nats.Msg, fn func(*In, string) (*Out, e _ = msg.Respond([]byte(`{"error":"unmarshal"}`)) return } - accountID := msg.Header.Get("X-Account-ID") + accountID := msg.Header.Get(utils.AccountIDHeader) result, err := fn(&input, accountID) if err != nil { errResp, _ := json.Marshal(map[string]string{"error": err.Error()}) diff --git a/hive/handlers/ec2/image/service_impl.go b/hive/handlers/ec2/image/service_impl.go index 48c238a54..f0d91f924 100644 --- a/hive/handlers/ec2/image/service_impl.go +++ b/hive/handlers/ec2/image/service_impl.go @@ -163,7 +163,7 @@ func (s *ImageServiceImpl) DescribeImages(input *ec2.DescribeImagesInput, accoun // in ImageOwnerAlias. Pre-phase4 AMIs have non-account values like "self" // or "hive" and are treated as system/public images visible to all. amiOwner := amiMeta.ImageOwnerAlias - isSystemAMI := !isAccountID(amiOwner) + isSystemAMI := !utils.IsAccountID(amiOwner) // Visibility check: callers can only see their own AMIs and system AMIs. // This runs regardless of whether an owner filter is specified. @@ -202,7 +202,7 @@ func (s *ImageServiceImpl) DescribeImages(input *ec2.DescribeImagesInput, accoun // Resolve the OwnerId for the response. System AMIs use global account. ownerID := amiOwner if isSystemAMI { - ownerID = "000000000000" + ownerID = utils.GlobalAccountID } // Build EC2 Image from AMIMetadata @@ -593,20 +593,6 @@ func (s *ImageServiceImpl) putSnapshotMetadata(snapshotID, volumeID string, volu return err } -// isAccountID returns true if the string looks like a valid Hive account ID -// (12-digit zero-padded number). Pre-phase4 values like "self" or "hive" return false. -func isAccountID(s string) bool { - if len(s) != 12 { - return false - } - for _, c := range s { - if c < '0' || c > '9' { - return false - } - } - return true -} - func (s *ImageServiceImpl) CopyImage(input *ec2.CopyImageInput, accountID string) (*ec2.CopyImageOutput, error) { return nil, errors.New("CopyImage not yet implemented") } diff --git a/hive/handlers/ec2/image/service_nats.go b/hive/handlers/ec2/image/service_nats.go index b4733d3f8..b274fe3e3 100644 --- a/hive/handlers/ec2/image/service_nats.go +++ b/hive/handlers/ec2/image/service_nats.go @@ -11,12 +11,15 @@ import ( // NATSImageService handles image operations via NATS messaging type NATSImageService struct { - natsConn *nats.Conn + natsConn *nats.Conn + expectedNodes int } -// NewNATSImageService creates a new NATS-based image service -func NewNATSImageService(conn *nats.Conn) ImageService { - return &NATSImageService{natsConn: conn} +// NewNATSImageService creates a new NATS-based image service. +// expectedNodes is used by scatter-gather operations (e.g. CreateImage) to +// enable early exit once all nodes have responded. +func NewNATSImageService(conn *nats.Conn, expectedNodes int) ImageService { + return &NATSImageService{natsConn: conn, expectedNodes: expectedNodes} } func (s *NATSImageService) DescribeImages(input *ec2.DescribeImagesInput, accountID string) (*ec2.DescribeImagesOutput, error) { @@ -24,7 +27,7 @@ func (s *NATSImageService) DescribeImages(input *ec2.DescribeImagesInput, accoun } func (s *NATSImageService) CreateImage(input *ec2.CreateImageInput, accountID string) (*ec2.CreateImageOutput, error) { - return utils.NATSRequest[ec2.CreateImageOutput](s.natsConn, "ec2.CreateImage", input, 120*time.Second, accountID) + return utils.NATSScatterGather[ec2.CreateImageOutput](s.natsConn, "ec2.CreateImage", input, 120*time.Second, s.expectedNodes, accountID) } func (s *NATSImageService) CopyImage(input *ec2.CopyImageInput, accountID string) (*ec2.CopyImageOutput, error) { diff --git a/hive/handlers/ec2/snapshot/service_impl.go b/hive/handlers/ec2/snapshot/service_impl.go index bd3ff9285..84b4e92e9 100644 --- a/hive/handlers/ec2/snapshot/service_impl.go +++ b/hive/handlers/ec2/snapshot/service_impl.go @@ -65,7 +65,7 @@ func NewSnapshotServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*S return nil, nil, fmt.Errorf("failed to get JetStream context: %w", err) } - kv, err := getOrCreateKVBucket(js, KVBucketVolumeSnapshots, 10) + kv, err := utils.GetOrCreateKVBucket(js, KVBucketVolumeSnapshots, 10) if err != nil { return nil, nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketVolumeSnapshots, err) } @@ -94,20 +94,6 @@ func NewSnapshotServiceImplWithStore(cfg *config.Config, store objectstore.Objec return svc } -func getOrCreateKVBucket(js nats.JetStreamContext, bucketName string, history int) (nats.KeyValue, error) { - kv, err := js.CreateKeyValue(&nats.KeyValueConfig{ - Bucket: bucketName, - History: utils.SafeIntToUint8(history), - }) - if err != nil { - kv, err = js.KeyValue(bucketName) - if err != nil { - return nil, err - } - } - return kv, nil -} - // getSnapshotKey returns the S3 key for storing snapshot metadata. // Uses metadata.json to avoid conflicting with viperblock's config.json // which stores the SnapshotState (block map references, source volume, etc). diff --git a/hive/handlers/ec2/vpc/eni.go b/hive/handlers/ec2/vpc/eni.go index 3a6bb40ef..ec1bcb829 100644 --- a/hive/handlers/ec2/vpc/eni.go +++ b/hive/handlers/ec2/vpc/eni.go @@ -38,9 +38,6 @@ type ENIRecord struct { // CreateNetworkInterface creates a new ENI in the specified subnet func (s *VPCServiceImpl) CreateNetworkInterface(input *ec2.CreateNetworkInterfaceInput, accountID string) (*ec2.CreateNetworkInterfaceOutput, error) { - if err := s.ensureKVReady(); err != nil { - return nil, errors.New(awserrors.ErrorServerInternal) - } if input.SubnetId == nil || *input.SubnetId == "" { return nil, errors.New(awserrors.ErrorMissingParameter) } @@ -48,7 +45,7 @@ func (s *VPCServiceImpl) CreateNetworkInterface(input *ec2.CreateNetworkInterfac subnetId := *input.SubnetId // Verify subnet exists and belongs to this account - subnetEntry, err := s.subnetKV.Get(accountKey(accountID, subnetId)) + subnetEntry, err := s.subnetKV.Get(utils.AccountKey(accountID, subnetId)) if err != nil { return nil, errors.New(awserrors.ErrorInvalidSubnetIDNotFound) } @@ -108,7 +105,7 @@ func (s *VPCServiceImpl) CreateNetworkInterface(input *ec2.CreateNetworkInterfac if err != nil { return nil, fmt.Errorf("failed to marshal ENI record: %w", err) } - if _, err := s.eniKV.Put(accountKey(accountID, eniId), data); err != nil { + if _, err := s.eniKV.Put(utils.AccountKey(accountID, eniId), data); err != nil { return nil, errors.New(awserrors.ErrorServerInternal) } @@ -124,15 +121,12 @@ func (s *VPCServiceImpl) CreateNetworkInterface(input *ec2.CreateNetworkInterfac // DeleteNetworkInterface deletes an ENI func (s *VPCServiceImpl) DeleteNetworkInterface(input *ec2.DeleteNetworkInterfaceInput, accountID string) (*ec2.DeleteNetworkInterfaceOutput, error) { - if err := s.ensureKVReady(); err != nil { - return nil, errors.New(awserrors.ErrorServerInternal) - } if input.NetworkInterfaceId == nil || *input.NetworkInterfaceId == "" { return nil, errors.New(awserrors.ErrorMissingParameter) } eniId := *input.NetworkInterfaceId - key := accountKey(accountID, eniId) + key := utils.AccountKey(accountID, eniId) // Get the ENI record entry, err := s.eniKV.Get(key) @@ -169,9 +163,6 @@ func (s *VPCServiceImpl) DeleteNetworkInterface(input *ec2.DeleteNetworkInterfac // DescribeNetworkInterfaces lists ENIs with optional filters func (s *VPCServiceImpl) DescribeNetworkInterfaces(input *ec2.DescribeNetworkInterfacesInput, accountID string) (*ec2.DescribeNetworkInterfacesOutput, error) { - if err := s.ensureKVReady(); err != nil { - return nil, errors.New(awserrors.ErrorServerInternal) - } var enis []*ec2.NetworkInterface eniIDs := make(map[string]bool) @@ -264,10 +255,7 @@ func (s *VPCServiceImpl) DescribeNetworkInterfaces(input *ec2.DescribeNetworkInt // AttachENI marks an ENI as attached to an instance (internal use by RunInstances). // accountID scopes the lookup to the correct KV key. func (s *VPCServiceImpl) AttachENI(accountID, eniId, instanceId string, deviceIndex int64) (string, error) { - if err := s.ensureKVReady(); err != nil { - return "", errors.New(awserrors.ErrorServerInternal) - } - key := accountKey(accountID, eniId) + key := utils.AccountKey(accountID, eniId) entry, err := s.eniKV.Get(key) if err != nil { return "", errors.New(awserrors.ErrorInvalidNetworkInterfaceIDNotFound) @@ -303,10 +291,7 @@ func (s *VPCServiceImpl) AttachENI(accountID, eniId, instanceId string, deviceIn // DetachENI marks an ENI as detached from an instance (internal use by TerminateInstances). // accountID scopes the lookup to the correct KV key. func (s *VPCServiceImpl) DetachENI(accountID, eniId string) error { - if err := s.ensureKVReady(); err != nil { - return errors.New(awserrors.ErrorServerInternal) - } - key := accountKey(accountID, eniId) + key := utils.AccountKey(accountID, eniId) entry, err := s.eniKV.Get(key) if err != nil { return errors.New(awserrors.ErrorInvalidNetworkInterfaceIDNotFound) diff --git a/hive/handlers/ec2/vpc/ipam.go b/hive/handlers/ec2/vpc/ipam.go index 244622aae..b9a072352 100644 --- a/hive/handlers/ec2/vpc/ipam.go +++ b/hive/handlers/ec2/vpc/ipam.go @@ -8,6 +8,7 @@ import ( "math/big" "net" + "github.com/mulgadc/hive/hive/utils" "github.com/nats-io/nats.go" ) @@ -29,7 +30,7 @@ type IPAM struct { // NewIPAM creates a new IPAM instance backed by NATS JetStream KV. func NewIPAM(js nats.JetStreamContext) (*IPAM, error) { - kv, err := getOrCreateKVBucket(js, KVBucketIPAM, 5) + kv, err := utils.GetOrCreateKVBucket(js, KVBucketIPAM, 5) if err != nil { return nil, fmt.Errorf("failed to create IPAM KV bucket: %w", err) } diff --git a/hive/handlers/ec2/vpc/service_impl.go b/hive/handlers/ec2/vpc/service_impl.go index a2f6fe74d..b803e5014 100644 --- a/hive/handlers/ec2/vpc/service_impl.go +++ b/hive/handlers/ec2/vpc/service_impl.go @@ -62,13 +62,6 @@ type VPCServiceImpl struct { ipam *IPAM } -// NewVPCServiceImpl creates a new VPC service implementation (in-memory, no persistence) -func NewVPCServiceImpl(cfg *config.Config) *VPCServiceImpl { - return &VPCServiceImpl{ - config: cfg, - } -} - // NewVPCServiceImplWithNATS creates a VPC service with NATS JetStream for persistence func NewVPCServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*VPCServiceImpl, error) { js, err := natsConn.JetStream() @@ -76,22 +69,22 @@ func NewVPCServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*VPCSer return nil, fmt.Errorf("failed to get JetStream context: %w", err) } - vpcKV, err := getOrCreateKVBucket(js, KVBucketVPCs, 10) + vpcKV, err := utils.GetOrCreateKVBucket(js, KVBucketVPCs, 10) if err != nil { return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketVPCs, err) } - subnetKV, err := getOrCreateKVBucket(js, KVBucketSubnets, 10) + subnetKV, err := utils.GetOrCreateKVBucket(js, KVBucketSubnets, 10) if err != nil { return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketSubnets, err) } - vniKV, err := getOrCreateKVBucket(js, KVBucketVNICounter, 10) + vniKV, err := utils.GetOrCreateKVBucket(js, KVBucketVNICounter, 10) if err != nil { return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketVNICounter, err) } - eniKV, err := getOrCreateKVBucket(js, KVBucketENIs, 10) + eniKV, err := utils.GetOrCreateKVBucket(js, KVBucketENIs, 10) if err != nil { return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketENIs, err) } @@ -118,35 +111,6 @@ func NewVPCServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*VPCSer }, nil } -func getOrCreateKVBucket(js nats.JetStreamContext, bucketName string, history int) (nats.KeyValue, error) { - kv, err := js.CreateKeyValue(&nats.KeyValueConfig{ - Bucket: bucketName, - History: utils.SafeIntToUint8(history), - }) - if err != nil { - kv, err = js.KeyValue(bucketName) - if err != nil { - return nil, err - } - } - return kv, nil -} - -// errKVNotReady is returned when the VPC service was initialized without NATS KV -// (in-memory fallback mode). Callers receive an internal server error. -var errKVNotReady = errors.New("VPC KV stores not initialized (in-memory fallback mode)") - -// ensureKVReady returns an error if any required KV store is nil. -// This guards against nil pointer dereferences when the daemon fell back to -// NewVPCServiceImpl (no NATS) but still receives VPC requests via queue group. -func (s *VPCServiceImpl) ensureKVReady() error { - if s.vpcKV == nil || s.subnetKV == nil || s.vniKV == nil || s.eniKV == nil { - slog.Error("VPC service called without KV stores initialized") - return errKVNotReady - } - return nil -} - // nextVNI allocates the next VNI using atomic increment on the NATS KV counter func (s *VPCServiceImpl) nextVNI() (int64, error) { entry, err := s.vniKV.Get(vniCounterKey) @@ -177,16 +141,8 @@ func (s *VPCServiceImpl) nextVNI() (int64, error) { return current, nil } -// accountKey returns a KV key scoped to the given account: "{accountID}.{resourceID}". -func accountKey(accountID, resourceID string) string { - return accountID + "." + resourceID -} - // CreateVpc creates a new VPC func (s *VPCServiceImpl) CreateVpc(input *ec2.CreateVpcInput, accountID string) (*ec2.CreateVpcOutput, error) { - if err := s.ensureKVReady(); err != nil { - return nil, errors.New(awserrors.ErrorServerInternal) - } if input.CidrBlock == nil || *input.CidrBlock == "" { return nil, errors.New(awserrors.ErrorMissingParameter) } @@ -235,7 +191,7 @@ func (s *VPCServiceImpl) CreateVpc(input *ec2.CreateVpcInput, accountID string) if err != nil { return nil, fmt.Errorf("failed to marshal VPC record: %w", err) } - if _, err := s.vpcKV.Put(accountKey(accountID, vpcID), data); err != nil { + if _, err := s.vpcKV.Put(utils.AccountKey(accountID, vpcID), data); err != nil { return nil, errors.New(awserrors.ErrorServerInternal) } @@ -251,15 +207,12 @@ func (s *VPCServiceImpl) CreateVpc(input *ec2.CreateVpcInput, accountID string) // DeleteVpc deletes a VPC func (s *VPCServiceImpl) DeleteVpc(input *ec2.DeleteVpcInput, accountID string) (*ec2.DeleteVpcOutput, error) { - if err := s.ensureKVReady(); err != nil { - return nil, errors.New(awserrors.ErrorServerInternal) - } if input.VpcId == nil || *input.VpcId == "" { return nil, errors.New(awserrors.ErrorMissingParameter) } vpcID := *input.VpcId - key := accountKey(accountID, vpcID) + key := utils.AccountKey(accountID, vpcID) if _, err := s.vpcKV.Get(key); err != nil { return nil, errors.New(awserrors.ErrorInvalidVpcIDNotFound) @@ -302,9 +255,6 @@ func (s *VPCServiceImpl) DeleteVpc(input *ec2.DeleteVpcInput, accountID string) // DescribeVpcs describes VPCs func (s *VPCServiceImpl) DescribeVpcs(input *ec2.DescribeVpcsInput, accountID string) (*ec2.DescribeVpcsOutput, error) { - if err := s.ensureKVReady(); err != nil { - return nil, errors.New(awserrors.ErrorServerInternal) - } var vpcs []*ec2.Vpc vpcIDs := make(map[string]bool) @@ -344,6 +294,32 @@ func (s *VPCServiceImpl) DescribeVpcs(input *ec2.DescribeVpcsInput, accountID st vpcs = append(vpcs, s.vpcRecordToEC2(&record, accountID)) } + // Include global default VPCs (created at startup) for non-root accounts. + // This matches AWS behavior where all accounts have a default VPC. + if accountID != utils.GlobalAccountID { + globalPrefix := utils.GlobalAccountID + "." + for _, key := range keys { + if !strings.HasPrefix(key, globalPrefix) { + continue + } + entry, err := s.vpcKV.Get(key) + if err != nil { + continue + } + var record VPCRecord + if err := json.Unmarshal(entry.Value(), &record); err != nil { + continue + } + if !record.IsDefault { + continue + } + if len(vpcIDs) > 0 && !vpcIDs[record.VpcId] { + continue + } + vpcs = append(vpcs, s.vpcRecordToEC2(&record, accountID)) + } + } + // If specific VPC IDs were requested but not found, return error if len(vpcIDs) > 0 { found := make(map[string]bool) @@ -368,9 +344,6 @@ func (s *VPCServiceImpl) DescribeVpcs(input *ec2.DescribeVpcsInput, accountID st // CreateSubnet creates a new subnet within a VPC func (s *VPCServiceImpl) CreateSubnet(input *ec2.CreateSubnetInput, accountID string) (*ec2.CreateSubnetOutput, error) { - if err := s.ensureKVReady(); err != nil { - return nil, errors.New(awserrors.ErrorServerInternal) - } if input.VpcId == nil || *input.VpcId == "" { return nil, errors.New(awserrors.ErrorMissingParameter) } @@ -381,7 +354,7 @@ func (s *VPCServiceImpl) CreateSubnet(input *ec2.CreateSubnetInput, accountID st vpcID := *input.VpcId // Verify VPC exists and belongs to this account - vpcEntry, err := s.vpcKV.Get(accountKey(accountID, vpcID)) + vpcEntry, err := s.vpcKV.Get(utils.AccountKey(accountID, vpcID)) if err != nil { return nil, errors.New(awserrors.ErrorInvalidVpcIDNotFound) } @@ -483,7 +456,7 @@ func (s *VPCServiceImpl) CreateSubnet(input *ec2.CreateSubnetInput, accountID st if err != nil { return nil, fmt.Errorf("failed to marshal subnet record: %w", err) } - if _, err := s.subnetKV.Put(accountKey(accountID, subnetID), data); err != nil { + if _, err := s.subnetKV.Put(utils.AccountKey(accountID, subnetID), data); err != nil { return nil, errors.New(awserrors.ErrorServerInternal) } @@ -499,15 +472,12 @@ func (s *VPCServiceImpl) CreateSubnet(input *ec2.CreateSubnetInput, accountID st // DeleteSubnet deletes a subnet func (s *VPCServiceImpl) DeleteSubnet(input *ec2.DeleteSubnetInput, accountID string) (*ec2.DeleteSubnetOutput, error) { - if err := s.ensureKVReady(); err != nil { - return nil, errors.New(awserrors.ErrorServerInternal) - } if input.SubnetId == nil || *input.SubnetId == "" { return nil, errors.New(awserrors.ErrorMissingParameter) } subnetID := *input.SubnetId - key := accountKey(accountID, subnetID) + key := utils.AccountKey(accountID, subnetID) // Read subnet record before deletion (needed for vpcd event) subnetEntry, err := s.subnetKV.Get(key) @@ -531,9 +501,6 @@ func (s *VPCServiceImpl) DeleteSubnet(input *ec2.DeleteSubnetInput, accountID st // DescribeSubnets describes subnets func (s *VPCServiceImpl) DescribeSubnets(input *ec2.DescribeSubnetsInput, accountID string) (*ec2.DescribeSubnetsOutput, error) { - if err := s.ensureKVReady(); err != nil { - return nil, errors.New(awserrors.ErrorServerInternal) - } var subnets []*ec2.Subnet subnetIDs := make(map[string]bool) @@ -677,11 +644,6 @@ const ( DefaultSubnetCidr = "172.31.0.0/20" ) -// GlobalAccountID is used for resources created without an explicit account context -// (e.g., default VPC at daemon startup). Pre-existing resources with no accountID -// are treated as owned by this account. -const GlobalAccountID = "000000000000" - // EnsureDefaultVPC creates a default VPC and subnet if none exists for the given account. // This matches AWS behavior where a default VPC is present on account creation. // Safe to call multiple times — no-ops if a default VPC already exists. @@ -736,7 +698,7 @@ func (s *VPCServiceImpl) EnsureDefaultVPC(accountID string) error { if err != nil { return fmt.Errorf("marshal default VPC: %w", err) } - if _, err := s.vpcKV.Put(accountKey(accountID, vpcID), data); err != nil { + if _, err := s.vpcKV.Put(utils.AccountKey(accountID, vpcID), data); err != nil { return fmt.Errorf("store default VPC: %w", err) } @@ -765,7 +727,7 @@ func (s *VPCServiceImpl) EnsureDefaultVPC(accountID string) error { if err != nil { return fmt.Errorf("marshal default subnet: %w", err) } - if _, err := s.subnetKV.Put(accountKey(accountID, subnetID), data); err != nil { + if _, err := s.subnetKV.Put(utils.AccountKey(accountID, subnetID), data); err != nil { return fmt.Errorf("store default subnet: %w", err) } @@ -784,10 +746,6 @@ func (s *VPCServiceImpl) EnsureDefaultVPC(accountID string) error { // GetDefaultSubnet returns the default subnet for RunInstances when no SubnetId is specified. func (s *VPCServiceImpl) GetDefaultSubnet(accountID string) (*SubnetRecord, error) { - if s.subnetKV == nil { - return nil, fmt.Errorf("VPC service not initialized") - } - prefix := accountID + "." keys, err := s.subnetKV.Keys() if err != nil { diff --git a/hive/handlers/ec2/vpc/service_impl_test.go b/hive/handlers/ec2/vpc/service_impl_test.go index 9b0cf3f56..c4588acd4 100644 --- a/hive/handlers/ec2/vpc/service_impl_test.go +++ b/hive/handlers/ec2/vpc/service_impl_test.go @@ -642,74 +642,6 @@ func TestVpcCidrBlockAssociation(t *testing.T) { assert.Equal(t, "associated", *out.Vpc.CidrBlockAssociationSet[0].CidrBlockState.State) } -// --- KV nil guard tests (mulga-x83) --- - -func TestKVNilGuard_CreateVpc(t *testing.T) { - svc := NewVPCServiceImpl(nil) // in-memory fallback, all KV fields nil - _, err := svc.CreateVpc(&ec2.CreateVpcInput{CidrBlock: aws.String("10.0.0.0/16")}, testAccountID) - assert.Error(t, err) - assert.Contains(t, err.Error(), "ServerInternal") -} - -func TestKVNilGuard_DeleteVpc(t *testing.T) { - svc := NewVPCServiceImpl(nil) - _, err := svc.DeleteVpc(&ec2.DeleteVpcInput{VpcId: aws.String("vpc-123")}, testAccountID) - assert.Error(t, err) - assert.Contains(t, err.Error(), "ServerInternal") -} - -func TestKVNilGuard_DescribeVpcs(t *testing.T) { - svc := NewVPCServiceImpl(nil) - _, err := svc.DescribeVpcs(&ec2.DescribeVpcsInput{}, testAccountID) - assert.Error(t, err) - assert.Contains(t, err.Error(), "ServerInternal") -} - -func TestKVNilGuard_CreateSubnet(t *testing.T) { - svc := NewVPCServiceImpl(nil) - _, err := svc.CreateSubnet(&ec2.CreateSubnetInput{ - VpcId: aws.String("vpc-123"), - CidrBlock: aws.String("10.0.1.0/24"), - }, testAccountID) - assert.Error(t, err) - assert.Contains(t, err.Error(), "ServerInternal") -} - -func TestKVNilGuard_DeleteSubnet(t *testing.T) { - svc := NewVPCServiceImpl(nil) - _, err := svc.DeleteSubnet(&ec2.DeleteSubnetInput{SubnetId: aws.String("subnet-123")}, testAccountID) - assert.Error(t, err) - assert.Contains(t, err.Error(), "ServerInternal") -} - -func TestKVNilGuard_DescribeSubnets(t *testing.T) { - svc := NewVPCServiceImpl(nil) - _, err := svc.DescribeSubnets(&ec2.DescribeSubnetsInput{}, testAccountID) - assert.Error(t, err) - assert.Contains(t, err.Error(), "ServerInternal") -} - -func TestKVNilGuard_CreateNetworkInterface(t *testing.T) { - svc := NewVPCServiceImpl(nil) - _, err := svc.CreateNetworkInterface(&ec2.CreateNetworkInterfaceInput{SubnetId: aws.String("subnet-123")}, testAccountID) - assert.Error(t, err) - assert.Contains(t, err.Error(), "ServerInternal") -} - -func TestKVNilGuard_DeleteNetworkInterface(t *testing.T) { - svc := NewVPCServiceImpl(nil) - _, err := svc.DeleteNetworkInterface(&ec2.DeleteNetworkInterfaceInput{NetworkInterfaceId: aws.String("eni-123")}, testAccountID) - assert.Error(t, err) - assert.Contains(t, err.Error(), "ServerInternal") -} - -func TestKVNilGuard_DescribeNetworkInterfaces(t *testing.T) { - svc := NewVPCServiceImpl(nil) - _, err := svc.DescribeNetworkInterfaces(&ec2.DescribeNetworkInterfacesInput{}, testAccountID) - assert.Error(t, err) - assert.Contains(t, err.Error(), "ServerInternal") -} - // --- Filter tests --- func TestDescribeVpcs_NilFields(t *testing.T) { @@ -883,14 +815,6 @@ func TestEnsureDefaultVPC_WithConfigAZ(t *testing.T) { assert.Equal(t, "us-west-2b", *subDesc.Subnets[0].AvailabilityZone) } -func TestGetDefaultSubnet_NoKV(t *testing.T) { - // Service without KV should return error - svc := NewVPCServiceImpl(nil) - _, err := svc.GetDefaultSubnet(testAccountID) - assert.Error(t, err) - assert.Contains(t, err.Error(), "not initialized") -} - func TestCreateVpc_NormalizesNetworkCidr(t *testing.T) { svc := setupTestVPCService(t) // Pass a CIDR with host bits set — should be normalized to network address diff --git a/hive/handlers/iam/account_test.go b/hive/handlers/iam/account_test.go index f4be841bf..686f124c4 100644 --- a/hive/handlers/iam/account_test.go +++ b/hive/handlers/iam/account_test.go @@ -5,6 +5,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/iam" + "github.com/mulgadc/hive/hive/utils" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -229,28 +230,28 @@ func TestSeedRootUser_AccountScoped(t *testing.T) { err = svc.SeedRootUser(&BootstrapData{ AccessKeyID: "AKIAROOTEXAMPLE12345", EncryptedSecret: encryptedSecret, - AccountID: GlobalAccountID, + AccountID: utils.GlobalAccountID, }) require.NoError(t, err) // Root user stored at 000000000000:root - out, err := svc.GetUser(GlobalAccountID, &iam.GetUserInput{ + out, err := svc.GetUser(utils.GlobalAccountID, &iam.GetUserInput{ UserName: aws.String("root"), }) require.NoError(t, err) assert.Equal(t, "root", *out.User.UserName) - assert.Contains(t, *out.User.Arn, GlobalAccountID) + assert.Contains(t, *out.User.Arn, utils.GlobalAccountID) // Access key has correct AccountID ak, err := svc.LookupAccessKey("AKIAROOTEXAMPLE12345") require.NoError(t, err) - assert.Equal(t, GlobalAccountID, ak.AccountID) + assert.Equal(t, utils.GlobalAccountID, ak.AccountID) assert.Equal(t, "root", ak.UserName) // Global account record was created - account, err := svc.GetAccount(GlobalAccountID) + account, err := svc.GetAccount(utils.GlobalAccountID) require.NoError(t, err) - assert.Equal(t, GlobalAccountID, account.AccountID) + assert.Equal(t, utils.GlobalAccountID, account.AccountID) assert.Equal(t, "Global", account.AccountName) assert.Equal(t, "ACTIVE", account.Status) } diff --git a/hive/handlers/iam/service_impl.go b/hive/handlers/iam/service_impl.go index 5723467ad..c00d41b99 100644 --- a/hive/handlers/iam/service_impl.go +++ b/hive/handlers/iam/service_impl.go @@ -16,6 +16,7 @@ import ( "github.com/aws/aws-sdk-go/service/iam" "github.com/mulgadc/hive/hive/admin" "github.com/mulgadc/hive/hive/awserrors" + "github.com/mulgadc/hive/hive/utils" "github.com/nats-io/nats.go" ) @@ -26,8 +27,6 @@ const ( KVBucketAccounts = "hive-accounts" KVBucketAccountCounter = "hive-account-counter" - GlobalAccountID = "000000000000" - maxAccessKeysPerUser = 2 ) @@ -512,7 +511,7 @@ func (s *IAMServiceImpl) DecryptSecret(ciphertext string) (string, error) { func (s *IAMServiceImpl) SeedRootUser(data *BootstrapData) error { // Ensure the global account record exists globalAccount := Account{ - AccountID: GlobalAccountID, + AccountID: utils.GlobalAccountID, AccountName: "Global", Status: AccountStatusActive, CreatedAt: time.Now().UTC().Format(time.RFC3339), @@ -521,16 +520,16 @@ func (s *IAMServiceImpl) SeedRootUser(data *BootstrapData) error { if err != nil { return fmt.Errorf("marshal global account: %w", err) } - if _, err := s.accountsBucket.Create(GlobalAccountID, accountData); err != nil && !errors.Is(err, nats.ErrKeyExists) { + if _, err := s.accountsBucket.Create(utils.GlobalAccountID, accountData); err != nil && !errors.Is(err, nats.ErrKeyExists) { return fmt.Errorf("seed global account: %w", err) } - kvKey := GlobalAccountID + ".root" + kvKey := utils.GlobalAccountID + ".root" rootUser := User{ UserName: "root", UserID: "AIDAAAAAAAAAAAAAAAAA", - AccountID: GlobalAccountID, - ARN: fmt.Sprintf("arn:aws:iam::%s:root", GlobalAccountID), + AccountID: utils.GlobalAccountID, + ARN: fmt.Sprintf("arn:aws:iam::%s:root", utils.GlobalAccountID), Path: "/", CreatedAt: time.Now().UTC().Format(time.RFC3339), AccessKeys: []string{data.AccessKeyID}, @@ -558,7 +557,7 @@ func (s *IAMServiceImpl) SeedRootUser(data *BootstrapData) error { AccessKeyID: data.AccessKeyID, SecretAccessKey: data.EncryptedSecret, UserName: "root", - AccountID: GlobalAccountID, + AccountID: utils.GlobalAccountID, Status: AccessKeyStatusActive, CreatedAt: rootUser.CreatedAt, } @@ -577,7 +576,7 @@ func (s *IAMServiceImpl) SeedRootUser(data *BootstrapData) error { return fmt.Errorf("seed root access key: %w", err) } - slog.Info("Root IAM user seeded", "accountID", GlobalAccountID, "accessKeyID", data.AccessKeyID) + slog.Info("Root IAM user seeded", "accountID", utils.GlobalAccountID, "accessKeyID", data.AccessKeyID) return nil } diff --git a/hive/handlers/iam/service_impl_test.go b/hive/handlers/iam/service_impl_test.go index 812e70c9f..979346095 100644 --- a/hive/handlers/iam/service_impl_test.go +++ b/hive/handlers/iam/service_impl_test.go @@ -9,6 +9,7 @@ import ( "github.com/aws/aws-sdk-go/service/iam" "github.com/mulgadc/hive/hive/admin" "github.com/mulgadc/hive/hive/awserrors" + "github.com/mulgadc/hive/hive/utils" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" @@ -18,7 +19,7 @@ import ( // Interface compliance check var _ IAMService = (*IAMServiceImpl)(nil) -const testAccountID = GlobalAccountID +const testAccountID = utils.GlobalAccountID func setupTestIAMService(t *testing.T) *IAMServiceImpl { t.Helper() @@ -570,24 +571,24 @@ func TestSeedRootUser(t *testing.T) { err = svc.SeedRootUser(&BootstrapData{ AccessKeyID: "AKIAEXAMPLE123456789", EncryptedSecret: encryptedSecret, - AccountID: GlobalAccountID, + AccountID: utils.GlobalAccountID, }) require.NoError(t, err) // Verify root user exists at account-scoped key - out, err := svc.GetUser(GlobalAccountID, &iam.GetUserInput{ + out, err := svc.GetUser(utils.GlobalAccountID, &iam.GetUserInput{ UserName: aws.String("root"), }) require.NoError(t, err) assert.Equal(t, "root", *out.User.UserName) - assert.Contains(t, *out.User.Arn, GlobalAccountID) + assert.Contains(t, *out.User.Arn, utils.GlobalAccountID) assert.Contains(t, *out.User.Arn, "root") // Verify access key exists with AccountID ak, err := svc.LookupAccessKey("AKIAEXAMPLE123456789") require.NoError(t, err) assert.Equal(t, "root", ak.UserName) - assert.Equal(t, GlobalAccountID, ak.AccountID) + assert.Equal(t, utils.GlobalAccountID, ak.AccountID) assert.Equal(t, "Active", ak.Status) // Verify secret is decryptable @@ -596,9 +597,9 @@ func TestSeedRootUser(t *testing.T) { assert.Equal(t, "test-secret-key", decrypted) // Verify global account record was created - account, err := svc.GetAccount(GlobalAccountID) + account, err := svc.GetAccount(utils.GlobalAccountID) require.NoError(t, err) - assert.Equal(t, GlobalAccountID, account.AccountID) + assert.Equal(t, utils.GlobalAccountID, account.AccountID) assert.Equal(t, "Global", account.AccountName) assert.Equal(t, "ACTIVE", account.Status) } @@ -610,7 +611,7 @@ func TestSeedRootUser_Idempotent(t *testing.T) { data := &BootstrapData{ AccessKeyID: "AKIAEXAMPLE123456789", EncryptedSecret: encryptedSecret, - AccountID: GlobalAccountID, + AccountID: utils.GlobalAccountID, } // First call seeds @@ -622,7 +623,7 @@ func TestSeedRootUser_Idempotent(t *testing.T) { require.NoError(t, err) // Root user should still exist with original data - out, err := svc.GetUser(GlobalAccountID, &iam.GetUserInput{ + out, err := svc.GetUser(utils.GlobalAccountID, &iam.GetUserInput{ UserName: aws.String("root"), }) require.NoError(t, err) diff --git a/hive/services/vpcd/integration_test.go b/hive/services/vpcd/integration_test.go index 34cdf0824..7d1b391d2 100644 --- a/hive/services/vpcd/integration_test.go +++ b/hive/services/vpcd/integration_test.go @@ -9,6 +9,8 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" handlers_ec2_igw "github.com/mulgadc/hive/hive/handlers/ec2/igw" + handlers_ec2_vpc "github.com/mulgadc/hive/hive/handlers/ec2/vpc" + "github.com/mulgadc/hive/hive/utils" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" ) @@ -153,6 +155,19 @@ func TestIntegration_VPCLifecycle(t *testing.T) { } // === Phase 4: Create and Attach Internet Gateway === + // Create VPC KV bucket and register test VPC so IGW ownership checks pass + js, err := nc.JetStream() + if err != nil { + t.Fatalf("JetStream: %v", err) + } + vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1}) + if err != nil { + t.Fatalf("create VPC KV: %v", err) + } + if _, err := vpcKV.Put(utils.AccountKey(testIGWAccountID, "vpc-integ1"), []byte(`{"vpc_id":"vpc-integ1","state":"available"}`)); err != nil { + t.Fatalf("register test VPC: %v", err) + } + igwSvc, err := handlers_ec2_igw.NewIGWServiceImplWithNATS(nil, nc) if err != nil { t.Fatalf("create IGW service: %v", err) @@ -373,6 +388,19 @@ func TestIntegration_MultiSubnetWithIGW(t *testing.T) { t.Errorf("router ports = %d, want 3", len(router.Ports)) } + // Create VPC KV bucket and register test VPC so IGW ownership checks pass + js, err := nc.JetStream() + if err != nil { + t.Fatalf("JetStream: %v", err) + } + vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1}) + if err != nil { + t.Fatalf("create VPC KV: %v", err) + } + if _, err := vpcKV.Put(utils.AccountKey(testIGWAccountID, "vpc-multi"), []byte(`{"vpc_id":"vpc-multi","state":"available"}`)); err != nil { + t.Fatalf("register test VPC: %v", err) + } + // Attach IGW igwSvc, _ := handlers_ec2_igw.NewIGWServiceImplWithNATS(nil, nc) igwOut, _ := igwSvc.CreateInternetGateway(&ec2.CreateInternetGatewayInput{}, testIGWAccountID) @@ -459,6 +487,21 @@ func TestIntegration_IGWErrorPaths(t *testing.T) { } }() + // Create VPC KV bucket and register test VPCs so IGW ownership checks pass + js, err := nc.JetStream() + if err != nil { + t.Fatalf("JetStream: %v", err) + } + vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1}) + if err != nil { + t.Fatalf("create VPC KV: %v", err) + } + for _, vpcID := range []string{"vpc-err1", "vpc-err2"} { + if _, err := vpcKV.Put(utils.AccountKey(testIGWAccountID, vpcID), []byte(`{"vpc_id":"`+vpcID+`","state":"available"}`)); err != nil { + t.Fatalf("register test VPC: %v", err) + } + } + igwSvc, err := handlers_ec2_igw.NewIGWServiceImplWithNATS(nil, nc) if err != nil { t.Fatalf("create IGW service: %v", err) diff --git a/hive/utils/account.go b/hive/utils/account.go new file mode 100644 index 000000000..082926c54 --- /dev/null +++ b/hive/utils/account.go @@ -0,0 +1,39 @@ +package utils + +import "github.com/nats-io/nats.go" + +// GlobalAccountID is the root/system account. Pre-Phase4 resources are owned by this account. +const GlobalAccountID = "000000000000" + +// AccountKey returns a KV key scoped to an account: "{accountID}.{resourceID}". +func AccountKey(accountID, resourceID string) string { + return accountID + "." + resourceID +} + +// IsAccountID checks if a string is a valid 12-digit AWS account ID. +func IsAccountID(s string) bool { + if len(s) != 12 { + return false + } + for _, c := range s { + if c < '0' || c > '9' { + return false + } + } + return true +} + +// GetOrCreateKVBucket creates or retrieves a NATS KV bucket. +func GetOrCreateKVBucket(js nats.JetStreamContext, bucketName string, history int) (nats.KeyValue, error) { + kv, err := js.CreateKeyValue(&nats.KeyValueConfig{ + Bucket: bucketName, + History: SafeIntToUint8(history), + }) + if err != nil { + kv, err = js.KeyValue(bucketName) + if err != nil { + return nil, err + } + } + return kv, nil +} diff --git a/hive/utils/account_test.go b/hive/utils/account_test.go new file mode 100644 index 000000000..ea6aa4b2e --- /dev/null +++ b/hive/utils/account_test.go @@ -0,0 +1,54 @@ +package utils + +import "testing" + +func TestAccountKey(t *testing.T) { + tests := []struct { + accountID string + resourceID string + want string + }{ + {"000000000000", "vpc-123", "000000000000.vpc-123"}, + {"123456789012", "igw-abc", "123456789012.igw-abc"}, + {"", "vol-1", ".vol-1"}, + } + for _, tt := range tests { + got := AccountKey(tt.accountID, tt.resourceID) + if got != tt.want { + t.Errorf("AccountKey(%q, %q) = %q, want %q", tt.accountID, tt.resourceID, got, tt.want) + } + } +} + +func TestIsAccountID(t *testing.T) { + tests := []struct { + input string + want bool + }{ + {"000000000000", true}, + {"123456789012", true}, + {"999999999999", true}, + {"self", false}, + {"hive", false}, + {"", false}, + {"12345678901", false}, // 11 digits + {"1234567890123", false}, // 13 digits + {"12345678901a", false}, // non-digit + {"abcdefghijkl", false}, + } + for _, tt := range tests { + got := IsAccountID(tt.input) + if got != tt.want { + t.Errorf("IsAccountID(%q) = %v, want %v", tt.input, got, tt.want) + } + } +} + +func TestGlobalAccountID(t *testing.T) { + if GlobalAccountID != "000000000000" { + t.Errorf("GlobalAccountID = %q, want %q", GlobalAccountID, "000000000000") + } + if !IsAccountID(GlobalAccountID) { + t.Error("GlobalAccountID should be a valid account ID") + } +} diff --git a/hive/utils/nats.go b/hive/utils/nats.go index 80684e386..9f65e65a2 100644 --- a/hive/utils/nats.go +++ b/hive/utils/nats.go @@ -77,6 +77,80 @@ func NATSRequest[Out any](conn *nats.Conn, subject string, input any, timeout ti return &output, nil } +// NATSScatterGather publishes a fan-out request and collects responses from +// multiple nodes. It returns the first successful (non-error) response. Error +// payloads from individual nodes are skipped. If all responses are errors, the +// last error is returned. If no responses arrive before the deadline, a timeout +// error is returned. When expectedNodes > 0, collection exits early once that +// many responses have been received. +func NATSScatterGather[Out any](conn *nats.Conn, subject string, input any, timeout time.Duration, expectedNodes int, accountID string) (*Out, error) { + jsonData, err := json.Marshal(input) + if err != nil { + return nil, fmt.Errorf("failed to marshal input: %w", err) + } + + inbox := nats.NewInbox() + sub, err := conn.SubscribeSync(inbox) + if err != nil { + return nil, fmt.Errorf("failed to create inbox: %w", err) + } + defer sub.Unsubscribe() + + pubMsg := nats.NewMsg(subject) + pubMsg.Reply = inbox + pubMsg.Data = jsonData + pubMsg.Header.Set(AccountIDHeader, accountID) + if err := conn.PublishMsg(pubMsg); err != nil { + return nil, fmt.Errorf("failed to publish request: %w", err) + } + + deadline := time.Now().Add(timeout) + responsesReceived := 0 + var lastErr error + + for time.Now().Before(deadline) { + if expectedNodes > 0 && responsesReceived >= expectedNodes { + break + } + + remaining := time.Until(deadline) + if remaining <= 0 { + break + } + + msg, err := sub.NextMsg(remaining) + if err != nil { + if err == nats.ErrTimeout || err == nats.ErrNoResponders { + break + } + return nil, fmt.Errorf("scatter-gather receive error: %w", err) + } + + responsesReceived++ + + responseError, err := ValidateErrorPayload(msg.Data) + if err != nil { + slog.Debug("ScatterGather: skipping error response", "code", *responseError.Code, "subject", subject) + lastErr = errors.New(*responseError.Code) + continue + } + + var output Out + if err := json.Unmarshal(msg.Data, &output); err != nil { + slog.Debug("ScatterGather: skipping malformed response", "subject", subject, "err", err) + lastErr = fmt.Errorf("failed to unmarshal response: %w", err) + continue + } + + return &output, nil + } + + if lastErr != nil { + return nil, lastErr + } + return nil, fmt.Errorf("scatter-gather timeout: no responses received for %s", subject) +} + // AccountIDFromMsg extracts the caller's account ID from a NATS message header. // Returns the account ID, or empty string if the header is not set. func AccountIDFromMsg(msg *nats.Msg) string { diff --git a/hive/utils/nats_test.go b/hive/utils/nats_test.go index e772d06c3..3d9d437f4 100644 --- a/hive/utils/nats_test.go +++ b/hive/utils/nats_test.go @@ -239,6 +239,145 @@ func TestNATSRequest_MarshalError(t *testing.T) { // --- AccountIDFromMsg tests --- +// --- NATSScatterGather tests --- + +func TestNATSScatterGather_SuccessAmongErrors(t *testing.T) { + ns := startTestNATSServer(t) + + nc, err := nats.Connect(ns.ClientURL()) + require.NoError(t, err) + defer nc.Close() + + type Resp struct { + Value string `json:"value"` + } + + // Simulate 3 nodes: 2 return errors, 1 returns success + _, err = nc.Subscribe("test.scatter", func(msg *nats.Msg) { + // Simulate varying response times + resp := Resp{Value: "found"} + data, _ := json.Marshal(resp) + msg.Respond(data) + }) + require.NoError(t, err) + + // Two error responders (faster) + for range 2 { + _, err = nc.Subscribe("test.scatter", func(msg *nats.Msg) { + errPayload := GenerateErrorPayload("InvalidInstanceID.NotFound") + msg.Respond(errPayload) + }) + require.NoError(t, err) + } + + result, err := NATSScatterGather[Resp](nc, "test.scatter", struct{}{}, 3*time.Second, 3, "") + require.NoError(t, err) + assert.Equal(t, "found", result.Value) +} + +func TestNATSScatterGather_AllErrors(t *testing.T) { + ns := startTestNATSServer(t) + + nc, err := nats.Connect(ns.ClientURL()) + require.NoError(t, err) + defer nc.Close() + + type Resp struct{} + + // All 3 nodes return errors + for range 3 { + _, err = nc.Subscribe("test.scatter.allerr", func(msg *nats.Msg) { + errPayload := GenerateErrorPayload("InvalidInstanceID.NotFound") + msg.Respond(errPayload) + }) + require.NoError(t, err) + } + + _, err = NATSScatterGather[Resp](nc, "test.scatter.allerr", struct{}{}, 2*time.Second, 3, "") + assert.Error(t, err) + assert.Contains(t, err.Error(), "InvalidInstanceID.NotFound") +} + +func TestNATSScatterGather_Timeout(t *testing.T) { + ns := startTestNATSServer(t) + + nc, err := nats.Connect(ns.ClientURL()) + require.NoError(t, err) + defer nc.Close() + + type Resp struct{} + + // No subscribers — should return a timeout error + _, err = NATSScatterGather[Resp](nc, "test.scatter.timeout", struct{}{}, 200*time.Millisecond, 0, "") + assert.Error(t, err) + assert.Contains(t, err.Error(), "no responses received") +} + +func TestNATSScatterGather_SingleSuccess(t *testing.T) { + ns := startTestNATSServer(t) + + nc, err := nats.Connect(ns.ClientURL()) + require.NoError(t, err) + defer nc.Close() + + type Resp struct { + ID string `json:"id"` + } + + _, err = nc.Subscribe("test.scatter.single", func(msg *nats.Msg) { + resp := Resp{ID: "ami-123"} + data, _ := json.Marshal(resp) + msg.Respond(data) + }) + require.NoError(t, err) + + result, err := NATSScatterGather[Resp](nc, "test.scatter.single", struct{}{}, 2*time.Second, 1, "acct-123") + require.NoError(t, err) + assert.Equal(t, "ami-123", result.ID) +} + +func TestNATSScatterGather_MarshalError(t *testing.T) { + ns := startTestNATSServer(t) + + nc, err := nats.Connect(ns.ClientURL()) + require.NoError(t, err) + defer nc.Close() + + type Resp struct{} + _, err = NATSScatterGather[Resp](nc, "test.scatter.marshal", make(chan int), 2*time.Second, 0, "") + assert.Error(t, err) + assert.Contains(t, err.Error(), "failed to marshal input") +} + +func TestNATSScatterGather_EarlyExitOnSuccess(t *testing.T) { + ns := startTestNATSServer(t) + + nc, err := nats.Connect(ns.ClientURL()) + require.NoError(t, err) + defer nc.Close() + + type Resp struct { + Value string `json:"value"` + } + + // One fast success responder — should return immediately without waiting + _, err = nc.Subscribe("test.scatter.early", func(msg *nats.Msg) { + resp := Resp{Value: "quick"} + data, _ := json.Marshal(resp) + msg.Respond(data) + }) + require.NoError(t, err) + + start := time.Now() + result, err := NATSScatterGather[Resp](nc, "test.scatter.early", struct{}{}, 5*time.Second, 0, "") + elapsed := time.Since(start) + + require.NoError(t, err) + assert.Equal(t, "quick", result.Value) + // Should return well before the 5s timeout + assert.Less(t, elapsed, 2*time.Second) +} + func TestAccountIDFromMsg(t *testing.T) { msg := nats.NewMsg("test") msg.Header.Set(AccountIDHeader, "444455556666")