diff --git a/hive/daemon/daemon_handlers.go b/hive/daemon/daemon_handlers.go index 50dcf6246..77c8a1b7b 100644 --- a/hive/daemon/daemon_handlers.go +++ b/hive/daemon/daemon_handlers.go @@ -135,6 +135,16 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command qmp.Command, instance return } + // Check AZ compatibility — volume and instance must be in the same AZ + if volCfg.VolumeMetadata.AvailabilityZone != "" && d.config.AZ != "" && + volCfg.VolumeMetadata.AvailabilityZone != d.config.AZ { + slog.Error("AttachVolume: volume and instance are in different AZs", + "volumeId", volumeID, "volumeAZ", volCfg.VolumeMetadata.AvailabilityZone, + "instanceAZ", d.config.AZ) + respondWithError(awserrors.ErrorInvalidVolumeZoneMismatch) + return + } + // Determine device name if device == "" { d.Instances.Mu.Lock() @@ -545,6 +555,18 @@ func (d *Daemon) handleStopOrTerminateInstance(msg *nats.Msg, command qmp.Comman slog.Info(action+" instance", "id", command.ID) + // Check state validity before attempting transition — return the correct + // AWS error code when the instance is already stopped/terminated/etc. + d.Instances.Mu.Lock() + currentState := instance.Status + d.Instances.Mu.Unlock() + if !vm.IsValidTransition(currentState, initialState) { + slog.Warn("Instance in incorrect state for "+strings.ToLower(action), + "instanceId", instance.ID, "currentState", string(currentState)) + respondWithError(awserrors.ErrorIncorrectInstanceState) + return + } + // Transition to the initial transitional state if err := d.TransitionState(instance, initialState); err != nil { slog.Error("Failed to transition to "+string(initialState), "instanceId", instance.ID, "err", err) @@ -709,6 +731,31 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { return } + // Validate AMI exists before allocating resources + if runInstancesInput.ImageId != nil { + _, err := d.imageService.GetAMIConfig(*runInstancesInput.ImageId) + if err != nil { + slog.Error("handleEC2RunInstances AMI not found", "imageId", *runInstancesInput.ImageId, "err", err) + errResp = utils.GenerateErrorPayload(awserrors.ErrorInvalidAMIIDNotFound) + if err := msg.Respond(errResp); err != nil { + slog.Error("Failed to respond to NATS request", "err", err) + } + return + } + } + + // Validate key pair exists (if specified) + if runInstancesInput.KeyName != nil && *runInstancesInput.KeyName != "" { + if err := d.keyService.ValidateKeyPairExists(*runInstancesInput.KeyName); err != nil { + slog.Error("handleEC2RunInstances key pair not found", "keyName", *runInstancesInput.KeyName, "err", err) + errResp = utils.GenerateErrorPayload(awserrors.ErrorInvalidKeyPairNotFound) + if err := msg.Respond(errResp); err != nil { + slog.Error("Failed to respond to NATS request", "err", err) + } + return + } + } + // Determine how many instances to launch based on MinCount/MaxCount minCount := int(*runInstancesInput.MinCount) maxCount := int(*runInstancesInput.MaxCount) @@ -769,11 +816,13 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { // Create all instances var instances []*vm.VM var allEC2Instances []*ec2.Instance + var lastRunErr error for i := 0; i < launchCount; i++ { instance, ec2Instance, err := d.instanceService.RunInstance(runInstancesInput) if err != nil { slog.Error("handleEC2RunInstances service.RunInstance failed", "index", i, "err", err) + lastRunErr = err // Deallocate this instance's resources d.resourceMgr.deallocate(instanceType) continue @@ -789,8 +838,15 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { for range instances { d.resourceMgr.deallocate(instanceType) } - slog.Error("handleEC2RunInstances failed to create minimum instances", "created", len(instances), "minCount", minCount) - errResp = utils.GenerateErrorPayload(awserrors.ErrorServerInternal) + // Propagate the service-layer error if it's a known AWS error code + errCode := awserrors.ErrorServerInternal + if lastRunErr != nil { + if _, ok := awserrors.ErrorLookup[lastRunErr.Error()]; ok { + errCode = lastRunErr.Error() + } + } + slog.Error("handleEC2RunInstances failed to create minimum instances", "created", len(instances), "minCount", minCount, "err", errCode) + errResp = utils.GenerateErrorPayload(errCode) if err := msg.Respond(errResp); err != nil { slog.Error("Failed to respond to NATS request", "err", err) } @@ -1194,11 +1250,17 @@ func (d *Daemon) handleEC2DescribeInstances(msg *nats.Msg) { d.Instances.Mu.Lock() defer d.Instances.Mu.Unlock() - // Filter instances if specific instance IDs were requested + // Validate and filter instances if specific instance IDs were requested instanceIDFilter := make(map[string]bool) if len(describeInstancesInput.InstanceIds) > 0 { for _, id := range describeInstancesInput.InstanceIds { - if id != nil { + if id != nil && *id != "" { + if !strings.HasPrefix(*id, "i-") { + if err := msg.Respond(utils.GenerateErrorPayload(awserrors.ErrorInvalidInstanceIDMalformed)); err != nil { + slog.Error("Failed to respond to NATS request", "err", err) + } + return + } instanceIDFilter[*id] = true } } diff --git a/hive/daemon/daemon_handlers_test.go b/hive/daemon/daemon_handlers_test.go index 85c9fcbde..9915548d8 100644 --- a/hive/daemon/daemon_handlers_test.go +++ b/hive/daemon/daemon_handlers_test.go @@ -5,15 +5,19 @@ import ( "encoding/json" "fmt" "os" + "strings" "testing" "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" + awss3 "github.com/aws/aws-sdk-go/service/s3" + "github.com/mulgadc/hive/hive/config" handlers_ec2_account "github.com/mulgadc/hive/hive/handlers/ec2/account" handlers_ec2_eigw "github.com/mulgadc/hive/hive/handlers/ec2/eigw" handlers_ec2_image "github.com/mulgadc/hive/hive/handlers/ec2/image" + handlers_ec2_instance "github.com/mulgadc/hive/hive/handlers/ec2/instance" handlers_ec2_key "github.com/mulgadc/hive/hive/handlers/ec2/key" handlers_ec2_snapshot "github.com/mulgadc/hive/hive/handlers/ec2/snapshot" handlers_ec2_tags "github.com/mulgadc/hive/hive/handlers/ec2/tags" @@ -21,14 +25,15 @@ import ( "github.com/mulgadc/hive/hive/objectstore" "github.com/mulgadc/hive/hive/qmp" "github.com/mulgadc/hive/hive/vm" + "github.com/mulgadc/viperblock/viperblock" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) -// createFullTestDaemon creates a test daemon with ALL services initialized (including -// key, image, snapshot, tags, eigw, account) using in-memory object stores. -func createFullTestDaemon(t *testing.T, natsURL string) *Daemon { +// createFullTestDaemonWithStore creates a test daemon with ALL services initialized +// and returns the shared memory store for seeding test data. +func createFullTestDaemonWithStore(t *testing.T, natsURL string) (*Daemon, *objectstore.MemoryObjectStore) { daemon := createTestDaemon(t, natsURL) memStore := objectstore.NewMemoryObjectStore() @@ -42,6 +47,13 @@ func createFullTestDaemon(t *testing.T, natsURL string) *Daemon { daemon.eigwService = handlers_ec2_eigw.NewEgressOnlyIGWServiceImpl(cfg) daemon.accountService = handlers_ec2_account.NewAccountSettingsServiceImpl(cfg) + return daemon, memStore +} + +// createFullTestDaemon creates a test daemon with ALL services initialized (including +// key, image, snapshot, tags, eigw, account) using in-memory object stores. +func createFullTestDaemon(t *testing.T, natsURL string) *Daemon { + daemon, _ := createFullTestDaemonWithStore(t, natsURL) return daemon } @@ -309,6 +321,164 @@ func TestHandleNodeDiscover(t *testing.T) { assert.Equal(t, daemon.node, resp.Node) } +// --- handleEC2RunInstances AMI validation tests --- + +func TestHandleEC2RunInstances_InvalidAMI(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createFullTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.QueueSubscribe("ec2.RunInstances", "hive-workers", daemon.handleEC2RunInstances) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.RunInstancesInput{ + ImageId: aws.String("ami-nonexistent"), + InstanceType: aws.String(getTestInstanceType()), + MinCount: aws.Int64(1), + MaxCount: aws.Int64(1), + } + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.RunInstances", reqData, 5*time.Second) + require.NoError(t, err) + + // Should return InvalidAMIID.NotFound, not ServerInternal + assert.Contains(t, string(reply.Data), "InvalidAMIID.NotFound") +} + +func TestHandleEC2RunInstances_InvalidKeyPair(t *testing.T) { + natsURL := sharedNATSURL + + daemon, memStore := createFullTestDaemonWithStore(t, natsURL) + + // Seed a valid AMI so AMI validation passes + seedTestAMI(t, memStore, daemon.config.Predastore.Bucket, "ami-test123") + + sub, err := daemon.natsConn.QueueSubscribe("ec2.RunInstances", "hive-workers", daemon.handleEC2RunInstances) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.RunInstancesInput{ + ImageId: aws.String("ami-test123"), + InstanceType: aws.String(getTestInstanceType()), + KeyName: aws.String("nonexistent-keypair"), + MinCount: aws.Int64(1), + MaxCount: aws.Int64(1), + } + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.RunInstances", reqData, 5*time.Second) + require.NoError(t, err) + + // Should return InvalidKeyPair.NotFound, not proceed to launch + assert.Contains(t, string(reply.Data), "InvalidKeyPair.NotFound") +} + +func TestHandleEC2RunInstances_ValidKeyPairPassesValidation(t *testing.T) { + natsURL := sharedNATSURL + + daemon, memStore := createFullTestDaemonWithStore(t, natsURL) + + // Seed a valid AMI + seedTestAMI(t, memStore, daemon.config.Predastore.Bucket, "ami-test456") + + // Seed a valid key pair (public key + metadata) + bucket := daemon.config.Predastore.Bucket + _, err := memStore.PutObject(&awss3.PutObjectInput{ + Bucket: aws.String(bucket), + Key: aws.String("keys/123456789/my-key"), + Body: strings.NewReader("ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAITest"), + }) + require.NoError(t, err) + + metadataJSON := `{"KeyPairId":"key-abc123","KeyName":"my-key","KeyFingerprint":"SHA256:test"}` + _, err = memStore.PutObject(&awss3.PutObjectInput{ + Bucket: aws.String(bucket), + Key: aws.String("keys/123456789/key-abc123.json"), + Body: strings.NewReader(metadataJSON), + }) + require.NoError(t, err) + + sub, err := daemon.natsConn.QueueSubscribe("ec2.RunInstances", "hive-workers", daemon.handleEC2RunInstances) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.RunInstancesInput{ + ImageId: aws.String("ami-test456"), + InstanceType: aws.String(getTestInstanceType()), + KeyName: aws.String("my-key"), + MinCount: aws.Int64(1), + MaxCount: aws.Int64(1), + } + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.RunInstances", reqData, 5*time.Second) + require.NoError(t, err) + + // Should NOT contain InvalidKeyPair.NotFound — key pair validation should pass + assert.NotContains(t, string(reply.Data), "InvalidKeyPair.NotFound") +} + +func TestHandleEC2RunInstances_EmptyKeyNameSkipsValidation(t *testing.T) { + natsURL := sharedNATSURL + + daemon, memStore := createFullTestDaemonWithStore(t, natsURL) + seedTestAMI(t, memStore, daemon.config.Predastore.Bucket, "ami-test789") + + sub, err := daemon.natsConn.QueueSubscribe("ec2.RunInstances", "hive-workers", daemon.handleEC2RunInstances) + require.NoError(t, err) + defer sub.Unsubscribe() + + // No KeyName at all — should skip validation + input := &ec2.RunInstancesInput{ + ImageId: aws.String("ami-test789"), + InstanceType: aws.String(getTestInstanceType()), + MinCount: aws.Int64(1), + MaxCount: aws.Int64(1), + } + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.RunInstances", reqData, 5*time.Second) + require.NoError(t, err) + + // Should NOT contain InvalidKeyPair.NotFound + assert.NotContains(t, string(reply.Data), "InvalidKeyPair.NotFound") +} + +// --- handleEC2RunInstances service-layer error propagation --- + +func TestHandleEC2RunInstances_ServiceErrorPropagated(t *testing.T) { + natsURL := sharedNATSURL + + daemon, memStore := createFullTestDaemonWithStore(t, natsURL) + seedTestAMI(t, memStore, daemon.config.Predastore.Bucket, "ami-propatest") + + // Override instanceService with one that has an empty instance types map. + // The resourceMgr still has instance types, so the daemon-level check passes, + // but RunInstance() will fail with ErrorInvalidInstanceType. + emptyTypes := map[string]*ec2.InstanceTypeInfo{} + daemon.instanceService = handlers_ec2_instance.NewInstanceServiceImpl( + daemon.config, emptyTypes, daemon.natsConn, &daemon.Instances, + objectstore.NewMemoryObjectStore(), + ) + + sub, err := daemon.natsConn.QueueSubscribe("ec2.RunInstances", "hive-workers", daemon.handleEC2RunInstances) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.RunInstancesInput{ + ImageId: aws.String("ami-propatest"), + InstanceType: aws.String(getTestInstanceType()), + MinCount: aws.Int64(1), + MaxCount: aws.Int64(1), + } + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.RunInstances", reqData, 5*time.Second) + require.NoError(t, err) + + // Should propagate the specific AWS error from the service layer, + // not swallow it into ServerInternal + assert.Contains(t, string(reply.Data), "InvalidInstanceType") + assert.NotContains(t, string(reply.Data), "ServerInternal") +} + // --- handleStopOrTerminateInstance tests (JetStream required for TransitionState) --- func TestHandleEC2Events_StopInstance(t *testing.T) { @@ -1204,3 +1374,73 @@ func TestHandleEC2GetConsoleOutput_NotFound(t *testing.T) { // Should get an error response (instance not found) assert.Contains(t, string(reply.Data), "InvalidInstanceID.NotFound") } + +// TestAttachVolume_ZoneMismatch verifies that attaching a volume in a different AZ +// returns InvalidVolume.ZoneMismatch instead of proceeding. +func TestAttachVolume_ZoneMismatch(t *testing.T) { + natsURL := sharedNATSURL + daemon, store := createFullTestDaemonWithStore(t, natsURL) + + // Set the daemon's AZ + daemon.config.AZ = "us-east-1a" + + instanceID := "i-test-az-mismatch" + volumeID := "vol-az-mismatch" + + // Create a running instance + instance := &vm.VM{ + ID: instanceID, + InstanceType: getTestInstanceType(), + Status: vm.StateRunning, + Instance: &ec2.Instance{}, + QMPClient: &qmp.QMPClient{}, + } + daemon.Instances.VMS[instanceID] = instance + + // Create a volume in a different AZ + wrapper := struct { + VolumeConfig viperblock.VolumeConfig `json:"VolumeConfig"` + }{ + VolumeConfig: viperblock.VolumeConfig{ + VolumeMetadata: viperblock.VolumeMetadata{ + VolumeID: volumeID, + SizeGiB: 10, + State: "available", + AvailabilityZone: "us-west-2a", + }, + }, + } + data, _ := json.Marshal(wrapper) + store.PutObject(&awss3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String(volumeID + "/config.json"), + Body: strings.NewReader(string(data)), + }) + + // Subscribe handler + sub, err := daemon.natsConn.Subscribe( + fmt.Sprintf("ec2.cmd.%s", instanceID), + daemon.handleEC2Events, + ) + require.NoError(t, err) + defer sub.Unsubscribe() + + command := qmp.Command{ + ID: instanceID, + Attributes: qmp.Attributes{ + AttachVolume: true, + }, + AttachVolumeData: &qmp.AttachVolumeData{ + VolumeID: volumeID, + }, + } + cmdData, _ := json.Marshal(command) + + resp, err := daemon.natsConn.Request( + fmt.Sprintf("ec2.cmd.%s", instanceID), + cmdData, + 5*time.Second, + ) + require.NoError(t, err) + assert.Contains(t, string(resp.Data), "InvalidVolume.ZoneMismatch") +} diff --git a/hive/daemon/daemon_test.go b/hive/daemon/daemon_test.go index 18316d8e6..7b573b293 100644 --- a/hive/daemon/daemon_test.go +++ b/hive/daemon/daemon_test.go @@ -14,6 +14,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" + awss3 "github.com/aws/aws-sdk-go/service/s3" "github.com/mulgadc/hive/hive/config" handlers_ec2_instance "github.com/mulgadc/hive/hive/handlers/ec2/instance" handlers_ec2_volume "github.com/mulgadc/hive/hive/handlers/ec2/volume" @@ -103,6 +104,20 @@ func getTestInstanceType() string { return "t3.micro" // Default fallback } +// seedTestAMI creates a minimal AMI config in the memory store so that +// handleEC2RunInstances AMI validation passes. +func seedTestAMI(t *testing.T, store *objectstore.MemoryObjectStore, bucket, imageID string) { + t.Helper() + amiConfig := `{"VolumeConfig":{"AMIMetadata":{"ImageID":"` + imageID + `","Name":"test"}}}` + _, err := store.PutObject(&awss3.PutObjectInput{ + Bucket: aws.String(bucket), + Key: aws.String(imageID + "/config.json"), + Body: strings.NewReader(amiConfig), + ContentType: aws.String("application/json"), + }) + require.NoError(t, err) +} + // TestHandleEC2RunInstances_MessageParsing tests that the handler correctly parses NATS messages func TestHandleEC2RunInstances_MessageParsing(t *testing.T) { // Skip integration tests on macOS since viperblock/nbdkit are not available @@ -1285,7 +1300,11 @@ func TestRunInstances_CountValidation(t *testing.T) { instanceType := getTestInstanceType() topic := fmt.Sprintf("ec2.RunInstances.%s", instanceType) - daemon := createTestDaemon(t, natsURL) + daemon, memStore := createFullTestDaemonWithStore(t, natsURL) + + // Seed a valid AMI in the store so tests that pass input validation + // don't fail at the AMI existence check + seedTestAMI(t, memStore, daemon.config.Predastore.Bucket, "ami-test") // Subscribe to the per-instance-type topic (matches production routing) sub, err := daemon.natsConn.QueueSubscribe(topic, "hive-workers", daemon.handleEC2RunInstances) @@ -3459,3 +3478,107 @@ func TestRollbackEBSMount_NATSTimeout(t *testing.T) { // Should not panic, just log the timeout daemon.rollbackEBSMount(ebsReq) } + +// TestDescribeInstances_InvalidInstanceIDMalformed verifies that DescribeInstances +// returns InvalidInstanceID.Malformed when given instance IDs without the i- prefix. +func TestDescribeInstances_InvalidInstanceIDMalformed(t *testing.T) { + natsURL := sharedNATSURL + daemon := createTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.Subscribe("ec2.DescribeInstances", daemon.handleEC2DescribeInstances) + require.NoError(t, err) + defer sub.Unsubscribe() + + t.Run("MalformedInstanceID", func(t *testing.T) { + input := &ec2.DescribeInstancesInput{ + InstanceIds: []*string{aws.String("bad-instance-id")}, + } + inputJSON, _ := json.Marshal(input) + + resp, err := daemon.natsConn.Request("ec2.DescribeInstances", inputJSON, 5*time.Second) + require.NoError(t, err) + assert.Contains(t, string(resp.Data), "InvalidInstanceID.Malformed") + }) + + t.Run("ValidInstanceIDPassesValidation", func(t *testing.T) { + input := &ec2.DescribeInstancesInput{ + InstanceIds: []*string{aws.String("i-nonexistent")}, + } + inputJSON, _ := json.Marshal(input) + + resp, err := daemon.natsConn.Request("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") + }) +} + +// TestStopTerminate_IncorrectInstanceState verifies that stopping an already-stopped +// instance or terminating an already-terminated instance returns IncorrectInstanceState +// instead of ServerInternal. +func TestStopTerminate_IncorrectInstanceState(t *testing.T) { + natsURL := sharedNATSURL + daemon := createTestDaemon(t, natsURL) + + instanceID := "i-test-state-check" + instance := &vm.VM{ + ID: instanceID, + InstanceType: getTestInstanceType(), + Status: vm.StateStopped, + Instance: &ec2.Instance{}, + } + daemon.Instances.VMS[instanceID] = instance + + sub, err := daemon.natsConn.Subscribe( + fmt.Sprintf("ec2.cmd.%s", instanceID), + daemon.handleEC2Events, + ) + require.NoError(t, err) + defer sub.Unsubscribe() + + t.Run("StopAlreadyStoppedInstance", func(t *testing.T) { + daemon.Instances.Mu.Lock() + instance.Status = vm.StateStopped + daemon.Instances.Mu.Unlock() + + command := qmp.Command{ + ID: instanceID, + Attributes: qmp.Attributes{ + StopInstance: true, + }, + } + data, _ := json.Marshal(command) + + resp, err := daemon.natsConn.Request( + fmt.Sprintf("ec2.cmd.%s", instanceID), + data, + 5*time.Second, + ) + require.NoError(t, err) + assert.Contains(t, string(resp.Data), "IncorrectInstanceState") + assert.NotContains(t, string(resp.Data), "ServerInternal") + }) + + t.Run("TerminateAlreadyTerminatedInstance", func(t *testing.T) { + daemon.Instances.Mu.Lock() + instance.Status = vm.StateTerminated + daemon.Instances.Mu.Unlock() + + command := qmp.Command{ + ID: instanceID, + Attributes: qmp.Attributes{ + TerminateInstance: true, + }, + } + data, _ := json.Marshal(command) + + resp, err := daemon.natsConn.Request( + fmt.Sprintf("ec2.cmd.%s", instanceID), + data, + 5*time.Second, + ) + require.NoError(t, err) + assert.Contains(t, string(resp.Data), "IncorrectInstanceState") + assert.NotContains(t, string(resp.Data), "ServerInternal") + }) +} diff --git a/hive/handlers/ec2/image/service_impl.go b/hive/handlers/ec2/image/service_impl.go index 86a17e888..3a2edfaf4 100644 --- a/hive/handlers/ec2/image/service_impl.go +++ b/hive/handlers/ec2/image/service_impl.go @@ -225,6 +225,21 @@ func (s *ImageServiceImpl) DescribeImages(input *ec2.DescribeImagesInput) (*ec2. images = append(images, image) } + // If specific ImageIds were requested, verify all were found + if len(input.ImageIds) > 0 { + foundIDs := make(map[string]bool, len(images)) + for _, img := range images { + if img.ImageId != nil { + foundIDs[*img.ImageId] = true + } + } + for _, reqID := range input.ImageIds { + if reqID != nil && !foundIDs[*reqID] { + return nil, errors.New(awserrors.ErrorInvalidAMIIDNotFound) + } + } + } + slog.Info("DescribeImages completed", "count", len(images)) return &ec2.DescribeImagesOutput{ @@ -246,6 +261,17 @@ func (s *ImageServiceImpl) CreateImageFromInstance(params CreateImageParams) (*e return nil, errors.New(awserrors.ErrorInvalidParameterValue) } + // Check for duplicate AMI name before doing any expensive work + name := aws.StringValue(input.Name) + if name != "" { + if exists, err := s.amiNameExists(name); err != nil { + slog.Error("CreateImageFromInstance: failed to check AMI name uniqueness", "name", name, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } else if exists { + return nil, errors.New(awserrors.ErrorInvalidAMINameDuplicate) + } + } + amiID := utils.GenerateResourceID("ami") snapshotID := utils.GenerateResourceID("snap") @@ -277,7 +303,7 @@ func (s *ImageServiceImpl) CreateImageFromInstance(params CreateImageParams) (*e Virtualization: "hvm", } if params.SourceImageID != "" { - srcCfg, err := s.getAMIConfig(params.SourceImageID) + srcCfg, err := s.GetAMIConfig(params.SourceImageID) if err != nil { slog.Error("CreateImageFromInstance: failed to read source AMI config", "sourceImageId", params.SourceImageID, "err", err) return nil, errors.New(awserrors.ErrorServerInternal) @@ -292,7 +318,6 @@ func (s *ImageServiceImpl) CreateImageFromInstance(params CreateImageParams) (*e } // Step 5: Build and store AMI config - name := aws.StringValue(input.Name) description := aws.StringValue(input.Description) amiConfig := viperblock.VBState{ @@ -453,8 +478,48 @@ func (s *ImageServiceImpl) getVolumeConfig(volumeID string) (*viperblock.VolumeC return &vbState.VolumeConfig, nil } -// getAMIConfig reads an AMI's metadata from S3 -func (s *ImageServiceImpl) getAMIConfig(imageID string) (viperblock.AMIMetadata, error) { +// amiNameExists checks if any existing AMI already uses the given name. +func (s *ImageServiceImpl) amiNameExists(name string) (bool, error) { + listResult, err := s.store.ListObjects(&s3.ListObjectsInput{ + Bucket: aws.String(s.bucketName), + Prefix: aws.String("ami-"), + Delimiter: aws.String("/"), + }) + if err != nil { + return false, fmt.Errorf("amiNameExists: failed to list AMIs: %w", err) + } + + for _, prefix := range listResult.CommonPrefixes { + if prefix.Prefix == nil { + continue + } + configKey := *prefix.Prefix + "config.json" + result, err := s.store.GetObject(&s3.GetObjectInput{ + Bucket: aws.String(s.bucketName), + Key: aws.String(configKey), + }) + if err != nil { + continue + } + + var vbState viperblock.VBState + decodeErr := json.NewDecoder(result.Body).Decode(&vbState) + _ = result.Body.Close() + if decodeErr != nil { + continue + } + + if vbState.VolumeConfig.AMIMetadata.Name == name { + return true, nil + } + } + + return false, nil +} + +// GetAMIConfig retrieves the AMI metadata for a given image ID from S3. +// Returns NoSuchKeyError if the AMI does not exist. +func (s *ImageServiceImpl) GetAMIConfig(imageID string) (viperblock.AMIMetadata, error) { configKey := fmt.Sprintf("%s/config.json", imageID) result, err := s.store.GetObject(&s3.GetObjectInput{ Bucket: aws.String(s.bucketName), diff --git a/hive/handlers/ec2/image/service_impl_test.go b/hive/handlers/ec2/image/service_impl_test.go index 9592929ee..2c8514ae7 100644 --- a/hive/handlers/ec2/image/service_impl_test.go +++ b/hive/handlers/ec2/image/service_impl_test.go @@ -74,6 +74,33 @@ func createTestAMIConfig(t *testing.T, store *objectstore.MemoryObjectStore, ima require.NoError(t, err) } +// createTestAMIConfigWithName creates a test AMI config with a specified name +func createTestAMIConfigWithName(t *testing.T, store *objectstore.MemoryObjectStore, imageID, name string) { + amiState := viperblock.VBState{ + VolumeConfig: viperblock.VolumeConfig{ + AMIMetadata: viperblock.AMIMetadata{ + ImageID: imageID, + Name: name, + Architecture: "x86_64", + PlatformDetails: "Linux/UNIX", + Virtualization: "hvm", + RootDeviceType: "ebs", + VolumeSizeGiB: 8, + }, + }, + } + data, err := json.Marshal(amiState) + require.NoError(t, err) + + _, err = store.PutObject(&awss3.PutObjectInput{ + Bucket: aws.String(testBucket), + Key: aws.String(imageID + "/config.json"), + Body: strings.NewReader(string(data)), + ContentType: aws.String("application/json"), + }) + require.NoError(t, err) +} + func TestCreateImageFromInstance_NilInput(t *testing.T) { svc, _ := setupTestImageService(t) @@ -162,7 +189,7 @@ func TestGetAMIConfig(t *testing.T) { createTestAMIConfig(t, store, "ami-abc123") - meta, err := svc.getAMIConfig("ami-abc123") + meta, err := svc.GetAMIConfig("ami-abc123") require.NoError(t, err) assert.Equal(t, "ami-abc123", meta.ImageID) assert.Equal(t, "test-ami", meta.Name) @@ -174,7 +201,7 @@ func TestGetAMIConfig(t *testing.T) { func TestGetAMIConfig_NotFound(t *testing.T) { svc, _ := setupTestImageService(t) - _, err := svc.getAMIConfig("ami-nonexistent") + _, err := svc.GetAMIConfig("ami-nonexistent") require.Error(t, err) } @@ -223,6 +250,37 @@ func TestCreateImageFromInstance_SourceAMIReadFailure(t *testing.T) { assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) } +func TestDescribeImages_NotFound(t *testing.T) { + svc, store := setupTestImageService(t) + + // Create one AMI + createTestAMIConfig(t, store, "ami-exists123") + + // Request a non-existent AMI ID — should return InvalidAMIID.NotFound + _, err := svc.DescribeImages(&ec2.DescribeImagesInput{ + ImageIds: []*string{aws.String("ami-nonexistent")}, + }) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidAMIIDNotFound, err.Error()) +} + +func TestDescribeImages_MixedExistingAndMissing(t *testing.T) { + svc, store := setupTestImageService(t) + + // Create one AMI + createTestAMIConfig(t, store, "ami-exists123") + + // Request one existing + one non-existent — should return NotFound + _, err := svc.DescribeImages(&ec2.DescribeImagesInput{ + ImageIds: []*string{ + aws.String("ami-exists123"), + aws.String("ami-missing456"), + }, + }) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidAMIIDNotFound, err.Error()) +} + func TestDescribeImages_FilterByOwnerSelf(t *testing.T) { svc, store := setupTestImageService(t) @@ -259,3 +317,49 @@ func TestDescribeImages_FilterByOwnerSelf(t *testing.T) { assert.Len(t, result.Images, 1) assert.Equal(t, amiID, *result.Images[0].ImageId) } + +func TestCreateImageFromInstance_DuplicateName(t *testing.T) { + svc, store := setupTestImageService(t) + + // Create an existing AMI with name "my-image" + createTestAMIConfigWithName(t, store, "ami-existing123", "my-image") + + // Create volume config for the snapshot step + createTestVolumeConfig(t, store, "vol-root123", 10) + + // Attempt to create another image with the same name — should fail with duplicate error + _, err := svc.CreateImageFromInstance(CreateImageParams{ + Input: &ec2.CreateImageInput{ + InstanceId: aws.String("i-test123"), + Name: aws.String("my-image"), + }, + RootVolumeID: "vol-root123", + IsRunning: true, + }) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidAMINameDuplicate, err.Error()) +} + +func TestCreateImageFromInstance_UniqueNameAllowed(t *testing.T) { + svc, store := setupTestImageService(t) + + // Create an existing AMI with a different name + createTestAMIConfigWithName(t, store, "ami-existing123", "other-image") + + // Create volume config + createTestVolumeConfig(t, store, "vol-root123", 10) + + // Creating with a unique name should NOT fail at the name check stage + // (it will fail later at the snapshot step since natsConn is nil, but that's expected) + _, err := svc.CreateImageFromInstance(CreateImageParams{ + Input: &ec2.CreateImageInput{ + InstanceId: aws.String("i-test123"), + Name: aws.String("unique-image"), + }, + RootVolumeID: "vol-root123", + IsRunning: true, + }) + require.Error(t, err) + // Should fail at snapshot, NOT at duplicate name check + assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) +} diff --git a/hive/handlers/ec2/instance/service_impl.go b/hive/handlers/ec2/instance/service_impl.go index 4e64f8c42..a3473bbab 100644 --- a/hive/handlers/ec2/instance/service_impl.go +++ b/hive/handlers/ec2/instance/service_impl.go @@ -318,8 +318,8 @@ func (s *InstanceServiceImpl) cloneAMIToVolume(input *ec2.RunInstancesInput, siz amiState, err := amiVb.LoadStateRequest("") if err != nil { - slog.Error("Could not load state for AMI", "err", err) - return errors.New(awserrors.ErrorServerInternal) + slog.Error("Could not load state for AMI", "imageId", *input.ImageId, "err", err) + return errors.New(awserrors.ErrorInvalidAMIIDNotFound) } snapshotID := amiState.VolumeConfig.AMIMetadata.SnapshotID @@ -546,6 +546,10 @@ func (s *InstanceServiceImpl) createCloudInitISO(input *ec2.RunInstancesInput, i Key: aws.String(keyPath), }) if err != nil { + if objectstore.IsNoSuchKeyError(err) { + slog.Error("key pair not found", "keyName", keyName, "err", err) + return errors.New(awserrors.ErrorInvalidKeyPairNotFound) + } slog.Error("failed to read SSH key", "err", err) return errors.New(awserrors.ErrorServerInternal) } diff --git a/hive/handlers/ec2/key/service_impl.go b/hive/handlers/ec2/key/service_impl.go index ab9ed10f6..6c049f9b7 100644 --- a/hive/handlers/ec2/key/service_impl.go +++ b/hive/handlers/ec2/key/service_impl.go @@ -362,6 +362,13 @@ func (s *KeyServiceImpl) findKeyPairIdFromKeyName(keyName string) (string, error return "", errors.New(awserrors.ErrorInvalidKeyPairNotFound) } +// ValidateKeyPairExists checks if a key pair with the given name exists. +// Returns nil if the key pair exists, or an error with ErrorInvalidKeyPairNotFound if not. +func (s *KeyServiceImpl) ValidateKeyPairExists(keyName string) error { + _, err := s.findKeyPairIdFromKeyName(keyName) + return err +} + // DeleteKeyPair removes a key pair (both public key and metadata from S3) func (s *KeyServiceImpl) DeleteKeyPair(input *ec2.DeleteKeyPairInput) (*ec2.DeleteKeyPairOutput, error) { if input == nil { @@ -386,6 +393,11 @@ func (s *KeyServiceImpl) DeleteKeyPair(input *ec2.DeleteKeyPairInput) (*ec2.Dele keyName, err = s.getKeyNameFromKeyPairId(keyPairID) if err != nil { + // AWS DeleteKeyPair is idempotent — return success for non-existent keys + if err.Error() == awserrors.ErrorInvalidKeyPairNotFound { + slog.Debug("DeleteKeyPair: key pair not found, returning success (idempotent)", "keyPairId", keyPairID) + return &ec2.DeleteKeyPairOutput{}, nil + } slog.Error("Failed to get keyName from keyPairId", "keyPairId", keyPairID, "err", err) return nil, err } @@ -401,6 +413,11 @@ func (s *KeyServiceImpl) DeleteKeyPair(input *ec2.DeleteKeyPairInput) (*ec2.Dele keyPairID, err = s.findKeyPairIdFromKeyName(keyName) if err != nil { + // AWS DeleteKeyPair is idempotent — return success for non-existent keys + if err.Error() == awserrors.ErrorInvalidKeyPairNotFound { + slog.Debug("DeleteKeyPair: key pair not found, returning success (idempotent)", "keyName", keyName) + return &ec2.DeleteKeyPairOutput{}, nil + } slog.Error("Failed to find keyPairId from keyName", "keyName", keyName, "err", err) return nil, err } @@ -596,7 +613,7 @@ func (s *KeyServiceImpl) ImportKeyPair(input *ec2.ImportKeyPairInput) (*ec2.Impo parts := strings.Fields(publicKeyString) if len(parts) < 2 { slog.Error("Invalid public key format", "keyName", keyName) - return nil, errors.New(awserrors.ErrorInvalidKeyPairFormat) + return nil, errors.New(awserrors.ErrorInvalidKeyFormat) } // Determine key type from algorithm prefix @@ -612,7 +629,13 @@ func (s *KeyServiceImpl) ImportKeyPair(input *ec2.ImportKeyPairInput) (*ec2.Impo keyType = "ecdsa" default: slog.Error("Unsupported key type", "algorithm", algorithmPrefix, "keyName", keyName) - return nil, errors.New(awserrors.ErrorInvalidKeyPairFormat) + return nil, errors.New(awserrors.ErrorInvalidKeyFormat) + } + + // Validate that the key data is valid base64 + if _, err := base64.StdEncoding.DecodeString(parts[1]); err != nil { + slog.Error("Invalid base64 in public key material", "keyName", keyName, "err", err) + return nil, errors.New(awserrors.ErrorInvalidKeyFormat) } // Calculate fingerprint from the imported public key diff --git a/hive/handlers/ec2/key/service_impl_test.go b/hive/handlers/ec2/key/service_impl_test.go index f350256eb..998f00a3b 100644 --- a/hive/handlers/ec2/key/service_impl_test.go +++ b/hive/handlers/ec2/key/service_impl_test.go @@ -7,6 +7,9 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/hive/hive/objectstore" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) // TestKeyPairMetadataMarshaling tests that CreateKeyPairOutput (used as metadata) can be marshaled/unmarshaled correctly @@ -389,6 +392,74 @@ func TestImportKeyPairKeyTypeDetection(t *testing.T) { } } +// TestDeleteKeyPairIdempotent verifies AWS-compatible idempotent behavior: +// deleting a non-existent key pair returns success instead of NotFound. +func TestDeleteKeyPairIdempotent(t *testing.T) { + store := objectstore.NewMemoryObjectStore() + svc := NewKeyServiceImplWithStore(store, "test-bucket", "123456789") + + t.Run("NonExistentKeyName", func(t *testing.T) { + result, err := svc.DeleteKeyPair(&ec2.DeleteKeyPairInput{ + KeyName: aws.String("no-such-key"), + }) + require.NoError(t, err) + assert.NotNil(t, result) + }) + + t.Run("NonExistentKeyPairId", func(t *testing.T) { + result, err := svc.DeleteKeyPair(&ec2.DeleteKeyPairInput{ + KeyPairId: aws.String("key-0123456789abcdef0"), + }) + require.NoError(t, err) + assert.NotNil(t, result) + }) +} + +// TestImportKeyPairInvalidKeyFormat verifies that ImportKeyPair returns +// InvalidKey.Format for malformed public key material (not InvalidKeyPair.Format). +func TestImportKeyPairInvalidKeyFormat(t *testing.T) { + store := objectstore.NewMemoryObjectStore() + svc := NewKeyServiceImplWithStore(store, "test-bucket", "123456789") + + tests := []struct { + name string + publicKey string + expectedErrMsg string + }{ + { + name: "SingleFieldNoKeyData", + publicKey: "ssh-rsa", + expectedErrMsg: "InvalidKey.Format", + }, + { + name: "UnsupportedAlgorithm", + publicKey: "ssh-dss AAAAB3NzaC1kc3MAAACB", + expectedErrMsg: "InvalidKey.Format", + }, + { + name: "InvalidBase64", + publicKey: "ssh-rsa not-valid-base64!!!", + expectedErrMsg: "InvalidKey.Format", + }, + { + name: "EmptyKeyData", + publicKey: "ssh-ed25519 ", + expectedErrMsg: "InvalidKey.Format", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := svc.ImportKeyPair(&ec2.ImportKeyPairInput{ + KeyName: aws.String("test-key"), + PublicKeyMaterial: []byte(tt.publicKey), + }) + require.Error(t, err) + assert.Equal(t, tt.expectedErrMsg, err.Error()) + }) + } +} + // TestImportKeyPairInputValidation tests input validation for ImportKeyPair func TestImportKeyPairInputValidation(t *testing.T) { tests := []struct { diff --git a/hive/handlers/ec2/snapshot/service_impl.go b/hive/handlers/ec2/snapshot/service_impl.go index 1f6312274..7c0a6cee5 100644 --- a/hive/handlers/ec2/snapshot/service_impl.go +++ b/hive/handlers/ec2/snapshot/service_impl.go @@ -354,6 +354,47 @@ func (s *SnapshotServiceImpl) DescribeSnapshots(input *ec2.DescribeSnapshotsInpu }, nil } +// snapshotInUseByVolumes checks if any volume was created from the given snapshot. +func (s *SnapshotServiceImpl) snapshotInUseByVolumes(snapshotID string) (bool, error) { + listResult, err := s.store.ListObjects(&s3.ListObjectsInput{ + Bucket: aws.String(s.config.Predastore.Bucket), + Prefix: aws.String("vol-"), + Delimiter: aws.String("/"), + }) + if err != nil { + return false, fmt.Errorf("snapshotInUseByVolumes: failed to list volumes: %w", err) + } + + for _, prefix := range listResult.CommonPrefixes { + if prefix.Prefix == nil { + continue + } + volumeID := strings.TrimSuffix(*prefix.Prefix, "/") + configKey := fmt.Sprintf("%s/config.json", volumeID) + + result, err := s.store.GetObject(&s3.GetObjectInput{ + Bucket: aws.String(s.config.Predastore.Bucket), + Key: aws.String(configKey), + }) + if err != nil { + continue // volume may not have a config yet + } + + var state viperblock.VBState + decodeErr := json.NewDecoder(result.Body).Decode(&state) + _ = result.Body.Close() + if decodeErr != nil { + continue + } + + if state.VolumeConfig.VolumeMetadata.SnapshotID == snapshotID { + return true, nil + } + } + + return false, nil +} + // DeleteSnapshot deletes a snapshot func (s *SnapshotServiceImpl) DeleteSnapshot(input *ec2.DeleteSnapshotInput) (*ec2.DeleteSnapshotOutput, error) { if input == nil || input.SnapshotId == nil { @@ -370,6 +411,17 @@ func (s *SnapshotServiceImpl) DeleteSnapshot(input *ec2.DeleteSnapshotInput) (*e return nil, err } + // Check if any volumes were created from this snapshot + inUse, err := s.snapshotInUseByVolumes(snapshotID) + if err != nil { + slog.Error("DeleteSnapshot failed to check snapshot usage", "snapshotId", snapshotID, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + if inUse { + slog.Info("DeleteSnapshot blocked: snapshot in use by volume", "snapshotId", snapshotID) + return nil, errors.New(awserrors.ErrorInvalidSnapshotInUse) + } + listResult, err := s.store.ListObjects(&s3.ListObjectsInput{ Bucket: aws.String(s.config.Predastore.Bucket), Prefix: aws.String(snapshotID + "/"), diff --git a/hive/handlers/ec2/snapshot/service_impl_test.go b/hive/handlers/ec2/snapshot/service_impl_test.go index 07e00426f..78b25c0bd 100644 --- a/hive/handlers/ec2/snapshot/service_impl_test.go +++ b/hive/handlers/ec2/snapshot/service_impl_test.go @@ -249,6 +249,50 @@ func TestDeleteSnapshot(t *testing.T) { assert.Empty(t, result.Snapshots) } +// TestDeleteSnapshot_InUseByVolume tests that deleting a snapshot fails when a volume was created from it +func TestDeleteSnapshot_InUseByVolume(t *testing.T) { + svc, store := setupTestSnapshotService(t) + + // Create a test volume and snapshot + createTestVolume(t, store, "vol-source", 50) + snap, err := svc.CreateSnapshot(&ec2.CreateSnapshotInput{ + VolumeId: aws.String("vol-source"), + }) + require.NoError(t, err) + + // Create a volume that references this snapshot (simulates CreateVolume from snapshot) + volumeState := viperblock.VBState{ + VolumeConfig: viperblock.VolumeConfig{ + VolumeMetadata: viperblock.VolumeMetadata{ + SizeGiB: 50, + SnapshotID: *snap.SnapshotId, + }, + }, + } + data, err := json.Marshal(volumeState) + require.NoError(t, err) + _, err = store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String("vol-cloned/config.json"), + Body: strings.NewReader(string(data)), + }) + require.NoError(t, err) + + // Attempt to delete the snapshot — should fail + _, err = svc.DeleteSnapshot(&ec2.DeleteSnapshotInput{ + SnapshotId: snap.SnapshotId, + }) + require.Error(t, err) + assert.Contains(t, err.Error(), awserrors.ErrorInvalidSnapshotInUse) + + // Verify snapshot still exists + result, err := svc.DescribeSnapshots(&ec2.DescribeSnapshotsInput{ + SnapshotIds: []*string{snap.SnapshotId}, + }) + require.NoError(t, err) + assert.Len(t, result.Snapshots, 1) +} + // TestDeleteSnapshot_NotFound tests deleting a non-existent snapshot func TestDeleteSnapshot_NotFound(t *testing.T) { svc, _ := setupTestSnapshotService(t) diff --git a/hive/handlers/ec2/volume/service_impl.go b/hive/handlers/ec2/volume/service_impl.go index 47b09e001..6d358cc48 100644 --- a/hive/handlers/ec2/volume/service_impl.go +++ b/hive/handlers/ec2/volume/service_impl.go @@ -234,7 +234,18 @@ func (s *VolumeServiceImpl) DescribeVolumes(input *ec2.DescribeVolumesInput) (*e // Fast path: if specific volume IDs are requested, fetch them directly if len(input.VolumeIds) > 0 { + // Count non-nil requested IDs + requested := 0 + for _, id := range input.VolumeIds { + if id != nil { + requested++ + } + } volumes = s.fetchVolumesByIDs(input.VolumeIds) + // AWS returns InvalidVolume.NotFound if any requested ID is missing + if len(volumes) != requested { + return nil, errors.New(awserrors.ErrorInvalidVolumeNotFound) + } slog.Info("DescribeVolumes completed", "count", len(volumes)) return &ec2.DescribeVolumesOutput{Volumes: volumes}, nil } diff --git a/hive/handlers/ec2/volume/service_impl_test.go b/hive/handlers/ec2/volume/service_impl_test.go index 2ab4036dc..d2e904ad4 100644 --- a/hive/handlers/ec2/volume/service_impl_test.go +++ b/hive/handlers/ec2/volume/service_impl_test.go @@ -784,12 +784,12 @@ func TestDescribeVolumes_FastPath_MixedExistingAndMissing(t *testing.T) { VolumeID: "vol-exists", SizeGiB: 10, State: "available", }) - output, err := svc.DescribeVolumes(&ec2.DescribeVolumesInput{ + // AWS returns InvalidVolume.NotFound when any requested ID is missing + _, err := svc.DescribeVolumes(&ec2.DescribeVolumesInput{ VolumeIds: []*string{aws.String("vol-exists"), aws.String("vol-ghost")}, }) - require.NoError(t, err) - assert.Len(t, output.Volumes, 1) - assert.Equal(t, "vol-exists", *output.Volumes[0].VolumeId) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidVolumeNotFound, err.Error()) } func TestDescribeVolumes_FastPath_NilVolumeID(t *testing.T) {