feat: EIGW (#33)

This commit is contained in:
Julian Sommer
2026-02-11 11:00:53 +11:00
committed by GitHub
parent 2937c8f425
commit e95c2536da
19 changed files with 1179 additions and 242 deletions
+194
View File
@@ -0,0 +1,194 @@
# Egress-Only Internet Gateway (EIGW)
## Overview
Egress-Only Internet Gateways provide outbound-only IPv6 connectivity for instances in a VPC. In Hive, EIGWs are lightweight metadata objects persisted in NATS JetStream KV. The implementation covers the three standard EC2 API operations: Create, Delete, and Describe.
## Key Design Decisions
### 1. NATS JetStream KV for persistence
EIGW records are stored in a dedicated KV bucket (`hive-eigw`) with 10-revision history. This avoids introducing a database dependency -- the same NATS infrastructure used for messaging also handles persistence.
The `getOrCreateKVBucket` helper tries `CreateKeyValue` first, then falls back to `KeyValue` (open existing). This makes the service idempotent across restarts.
**Trade-off**: KV is not suited for complex queries (e.g. filtering by tag values). Describe currently iterates all keys and filters in-memory, which is fine for typical EIGW counts but wouldn't scale to thousands of records.
### 2. Constructor fails loudly, daemon falls back gracefully
`NewEgressOnlyIGWServiceImplWithNATS` returns an error if the KV bucket can't be created or opened. This ensures the service never operates with a nil KV handle, eliminating an entire class of nil-pointer bugs.
The daemon handles this with a graceful fallback:
```go
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
}
```
**Trade-off**: The in-memory fallback (`NewEgressOnlyIGWServiceImpl`) has no KV backing, so EIGW state won't survive daemon restarts. This is acceptable for degraded mode -- the alternative is refusing to start the daemon entirely.
### 3. crypto/rand for ID generation
EIGW IDs (`eigw-` + 16 hex chars) are generated using `crypto/rand` + `encoding/hex`. This matches the pattern used by instance and volume ID generation elsewhere in the codebase.
```go
func generateEgressOnlyIGWID() string {
b := make([]byte, 8)
if _, err := rand.Read(b); err != nil {
panic("crypto/rand failed: " + err.Error())
}
return "eigw-" + hex.EncodeToString(b)
}
```
The panic on `crypto/rand` failure is intentional -- if the OS entropy source is broken, the system has larger problems.
### 4. Error strings are awserrors constants
All errors returned from `service_impl.go` use `errors.New(awserrors.ErrorXxx)` with exact constant values. This is critical because the error string travels through a round-trip:
```
service_impl: errors.New(awserrors.ErrorMissingParameter)
-> daemon handleNATSRequest: utils.GenerateErrorPayload(err.Error())
-> JSON over NATS: {"Code": "MissingParameter"}
-> gateway NATSRequest[Out]: ValidateErrorPayload extracts Code
-> gateway ErrorHandler: awserrors.ErrorLookup[err.Error()] -> HTTP status + XML
```
If the error string doesn't match a key in `ErrorLookup`, the gateway returns a generic 500. Using `fmt.Errorf` with extra context would break this chain.
### 5. Centralized nil NATS check in the gateway
Rather than checking `natsConn == nil` in each gateway function, a single check in `EC2_Request` covers all NATS-dependent actions. An allowlist (`ec2LocalActions`) exempts actions that work without NATS:
```go
var ec2LocalActions = map[string]bool{
"DescribeRegions": true,
"DescribeAvailabilityZones": true,
"DescribeAccountAttributes": true,
}
```
If `gw.NATSConn` is nil and the action isn't in the allowlist, the gateway returns `ErrorServerInternal` immediately.
### 6. Tag filtering by resource type
Create accepts `TagSpecifications` but only applies tags where `ResourceType` equals `"egress-only-internet-gateway"`. Tags with other resource types (e.g. `"instance"`) are silently ignored, matching AWS behavior.
## Request Flow
```
AWS SDK (create-egress-only-internet-gateway --vpc-id vpc-xxx)
-> Gateway (port 9999): ec2Handler parses AWS query params into SDK struct
-> Validate input (nil check, required fields)
-> NATS ec2.CreateEgressOnlyInternetGateway: routed to daemon via "hive-workers" queue group
-> Daemon handleEC2CreateEgressOnlyInternetGateway
-> handleNATSRequest: unmarshal -> EgressOnlyIGWServiceImpl.Create -> marshal -> respond
1. Validate VpcId is present
2. Generate eigw-{16hex} ID via crypto/rand
3. Build EgressOnlyIGWRecord with state "attached"
4. Process matching tag specifications
5. JSON marshal and store in NATS KV (hive-eigw bucket)
<- NATS response: ec2.CreateEgressOnlyInternetGatewayOutput JSON
<- Gateway: wrap in XML (CreateEgressOnlyInternetGatewayResponse)
<- AWS SDK: response
```
## NATS Topics
| Topic | Direction | Purpose |
|-------|-----------|---------|
| `ec2.CreateEgressOnlyInternetGateway` | Gateway -> Daemon | Create new EIGW attached to a VPC |
| `ec2.DeleteEgressOnlyInternetGateway` | Gateway -> Daemon | Delete EIGW by ID |
| `ec2.DescribeEgressOnlyInternetGateways` | Gateway -> Daemon | List/filter EIGWs |
All topics use the `hive-workers` queue group for load balancing across daemon instances.
## Storage Layout
```
NATS JetStream KV bucket: hive-eigw (history: 10)
Key: eigw-a1b2c3d4e5f6g7h8
Value: {
"egress_only_internet_gateway_id": "eigw-a1b2c3d4e5f6g7h8",
"vpc_id": "vpc-abc123",
"state": "attached",
"tags": {"Name": "my-eigw", "Env": "prod"},
"created_at": "2026-02-10T12:00:00Z"
}
```
## Files Changed
| File | Change |
|------|--------|
| `hive/handlers/ec2/eigw/service.go` | `EgressOnlyIGWService` interface (3 methods) |
| `hive/handlers/ec2/eigw/service_impl.go` | Direct implementation with NATS JetStream KV persistence |
| `hive/handlers/ec2/eigw/service_impl_test.go` | 12 test cases against embedded NATS server |
| `hive/handlers/ec2/eigw/service_nats.go` | NATS proxy (gateway -> daemon via `utils.NATSRequest`) |
| `hive/gateway/ec2/eigw/CreateEgressOnlyInternetGateway.go` | Gateway validation + handler for Create |
| `hive/gateway/ec2/eigw/DeleteEgressOnlyInternetGateway.go` | Gateway validation + handler for Delete |
| `hive/gateway/ec2/eigw/DescribeEgressOnlyInternetGateways.go` | Gateway validation + handler for Describe |
| `hive/gateway/ec2/eigw/eigw_test.go` | Gateway validation tests (9 cases) |
| `hive/gateway/ec2.go` | 3 entries in `ec2Actions` map; centralized nil NATS check |
| `hive/daemon/daemon.go` | `eigwService` field, 3 NATS subscriptions, service initialization with fallback |
| `hive/daemon/daemon_handlers.go` | 3 handler functions delegating to `handleNATSRequest` |
| `hive/awserrors/awserrors.go` | `ErrorInvalidEgressOnlyInternetGatewayId.Malformed` and `.NotFound` constants + lookup entries |
## Test Coverage
### Service Implementation Tests (`service_impl_test.go`)
| Test | What it validates |
|------|-------------------|
| `TestCreateEgressOnlyInternetGateway` | Basic create: ID format, VPC attachment, state |
| `TestCreateEgressOnlyInternetGateway_MissingVpcId` | Rejects nil VpcId with MissingParameter |
| `TestCreateEgressOnlyInternetGateway_EmptyVpcId` | Rejects empty string VpcId |
| `TestCreateEgressOnlyInternetGateway_WithTags` | Tags applied and persisted through describe round-trip |
| `TestCreateEgressOnlyInternetGateway_TagsWrongResourceType` | Tags with non-matching resource type are ignored |
| `TestDeleteEgressOnlyInternetGateway` | Delete succeeds, describe confirms removal |
| `TestDeleteEgressOnlyInternetGateway_NotFound` | Returns NotFound for nonexistent ID |
| `TestDeleteEgressOnlyInternetGateway_MissingID` | Rejects nil ID with MissingParameter |
| `TestDeleteEgressOnlyInternetGateway_EmptyID` | Rejects empty string ID |
| `TestDescribeEgressOnlyInternetGateways_All` | Lists all EIGWs when no filter specified |
| `TestDescribeEgressOnlyInternetGateways_ByID` | Filters by specific EIGW ID |
| `TestDescribeEgressOnlyInternetGateways_Empty` | Returns empty list for empty bucket |
All tests run against an embedded NATS server with JetStream enabled.
### Gateway Validation Tests (`eigw_test.go`)
| Test | What it validates |
|------|-------------------|
| `ValidateCreateEgressOnlyInternetGatewayInput` | nil input, missing VpcId, empty VpcId, valid input |
| `ValidateDeleteEgressOnlyInternetGatewayInput` | nil input, missing ID, empty ID, valid input |
| `ValidateDescribeEgressOnlyInternetGatewaysInput` | nil input, empty input, with filter IDs |
## AWS CLI Usage
```bash
# Create
aws ec2 create-egress-only-internet-gateway \
--vpc-id vpc-abc123 \
--tag-specifications 'ResourceType=egress-only-internet-gateway,Tags=[{Key=Name,Value=my-eigw}]' \
--endpoint-url https://localhost:9999
# Describe all
aws ec2 describe-egress-only-internet-gateways \
--endpoint-url https://localhost:9999
# Describe by ID
aws ec2 describe-egress-only-internet-gateways \
--egress-only-internet-gateway-ids eigw-a1b2c3d4e5f6g7h8 \
--endpoint-url https://localhost:9999
# Delete
aws ec2 delete-egress-only-internet-gateway \
--egress-only-internet-gateway-id eigw-a1b2c3d4e5f6g7h8 \
--endpoint-url https://localhost:9999
```
+11
View File
@@ -27,6 +27,7 @@ import (
"github.com/mulgadc/hive/hive/awserrors"
"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"
@@ -84,6 +85,7 @@ type Daemon struct {
accountService *handlers_ec2_account.AccountSettingsServiceImpl
snapshotService *handlers_ec2_snapshot.SnapshotServiceImpl
tagsService *handlers_ec2_tags.TagsServiceImpl
eigwService *handlers_ec2_eigw.EgressOnlyIGWServiceImpl
ctx context.Context
cancel context.CancelFunc
shutdownWg sync.WaitGroup
@@ -454,6 +456,9 @@ func (d *Daemon) subscribeAll() error {
{"ec2.CreateTags", d.handleEC2CreateTags, "hive-workers"},
{"ec2.DeleteTags", d.handleEC2DeleteTags, "hive-workers"},
{"ec2.DescribeTags", d.handleEC2DescribeTags, "hive-workers"},
{"ec2.CreateEgressOnlyInternetGateway", d.handleEC2CreateEgressOnlyInternetGateway, "hive-workers"},
{"ec2.DeleteEgressOnlyInternetGateway", d.handleEC2DeleteEgressOnlyInternetGateway, "hive-workers"},
{"ec2.DescribeEgressOnlyInternetGateways", d.handleEC2DescribeEgressOnlyInternetGateways, "hive-workers"},
{"ec2.DescribeInstances", d.handleEC2DescribeInstances, ""},
{"ec2.DescribeInstanceTypes", d.handleEC2DescribeInstanceTypes, ""},
{"ec2.startinstances", d.handleEC2StartInstances, "hive-workers"},
@@ -509,6 +514,12 @@ func (d *Daemon) Start() error {
d.snapshotService = handlers_ec2_snapshot.NewSnapshotServiceImpl(d.config, d.natsConn)
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
}
accountSvc, err := handlers_ec2_account.NewAccountSettingsServiceImplWithNATS(d.config, d.natsConn)
if err != nil {
slog.Warn("Failed to create account settings service with NATS, using in-memory fallback", "error", err)
+12 -1
View File
@@ -972,6 +972,18 @@ func (d *Daemon) handleEC2DescribeTags(msg *nats.Msg) {
handleNATSRequest(msg, d.tagsService.DescribeTags)
}
func (d *Daemon) handleEC2CreateEgressOnlyInternetGateway(msg *nats.Msg) {
handleNATSRequest(msg, d.eigwService.CreateEgressOnlyInternetGateway)
}
func (d *Daemon) handleEC2DeleteEgressOnlyInternetGateway(msg *nats.Msg) {
handleNATSRequest(msg, d.eigwService.DeleteEgressOnlyInternetGateway)
}
func (d *Daemon) handleEC2DescribeEgressOnlyInternetGateways(msg *nats.Msg) {
handleNATSRequest(msg, d.eigwService.DescribeEgressOnlyInternetGateways)
}
// handleEC2DescribeInstanceTypes processes incoming EC2 DescribeInstanceTypes requests
// This handler responds with instance types that can currently be provisioned on this node
// based on available resources (CPU and memory not already allocated to running instances)
@@ -1218,4 +1230,3 @@ func (d *Daemon) handleEC2EnableSerialConsoleAccess(msg *nats.Msg) {
func (d *Daemon) handleEC2DisableSerialConsoleAccess(msg *nats.Msg) {
handleNATSRequest(msg, d.accountService.DisableSerialConsoleAccess)
}
+22
View File
@@ -9,7 +9,9 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/gofiber/fiber/v2"
"github.com/mulgadc/hive/hive/awsec2query"
"github.com/mulgadc/hive/hive/awserrors"
gateway_ec2_account "github.com/mulgadc/hive/hive/gateway/ec2/account"
gateway_ec2_eigw "github.com/mulgadc/hive/hive/gateway/ec2/eigw"
gateway_ec2_image "github.com/mulgadc/hive/hive/gateway/ec2/image"
gateway_ec2_instance "github.com/mulgadc/hive/hive/gateway/ec2/instance"
gateway_ec2_key "github.com/mulgadc/hive/hive/gateway/ec2/key"
@@ -154,6 +156,22 @@ var ec2Actions = map[string]EC2Handler{
"CopySnapshot": ec2Handler(func(input *ec2.CopySnapshotInput, gw *GatewayConfig) (any, error) {
return gateway_ec2_snapshot.CopySnapshot(input, gw.NATSConn)
}),
"CreateEgressOnlyInternetGateway": ec2Handler(func(input *ec2.CreateEgressOnlyInternetGatewayInput, gw *GatewayConfig) (any, error) {
return gateway_ec2_eigw.CreateEgressOnlyInternetGateway(input, gw.NATSConn)
}),
"DeleteEgressOnlyInternetGateway": ec2Handler(func(input *ec2.DeleteEgressOnlyInternetGatewayInput, gw *GatewayConfig) (any, error) {
return gateway_ec2_eigw.DeleteEgressOnlyInternetGateway(input, gw.NATSConn)
}),
"DescribeEgressOnlyInternetGateways": ec2Handler(func(input *ec2.DescribeEgressOnlyInternetGatewaysInput, gw *GatewayConfig) (any, error) {
return gateway_ec2_eigw.DescribeEgressOnlyInternetGateways(input, gw.NATSConn)
}),
}
// ec2LocalActions are actions that don't require a NATS connection.
var ec2LocalActions = map[string]bool{
"DescribeRegions": true,
"DescribeAvailabilityZones": true,
"DescribeAccountAttributes": true,
}
func (gw *GatewayConfig) EC2_Request(ctx *fiber.Ctx) error {
@@ -165,6 +183,10 @@ func (gw *GatewayConfig) EC2_Request(ctx *fiber.Ctx) error {
return errors.New("InvalidAction")
}
if gw.NATSConn == nil && !ec2LocalActions[action] {
return errors.New(awserrors.ErrorServerInternal)
}
xmlOutput, err := handler(action, queryArgs, gw)
if err != nil {
return err
@@ -0,0 +1,40 @@
package gateway_ec2_eigw
import (
"errors"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/hive/hive/awserrors"
handlers_ec2_eigw "github.com/mulgadc/hive/hive/handlers/ec2/eigw"
"github.com/nats-io/nats.go"
)
// ValidateCreateEgressOnlyInternetGatewayInput validates the input parameters
func ValidateCreateEgressOnlyInternetGatewayInput(input *ec2.CreateEgressOnlyInternetGatewayInput) error {
if input == nil {
return errors.New(awserrors.ErrorInvalidParameterValue)
}
if input.VpcId == nil || *input.VpcId == "" {
return errors.New(awserrors.ErrorMissingParameter)
}
return nil
}
// CreateEgressOnlyInternetGateway handles the EC2 CreateEgressOnlyInternetGateway API call
func CreateEgressOnlyInternetGateway(input *ec2.CreateEgressOnlyInternetGatewayInput, natsConn *nats.Conn) (ec2.CreateEgressOnlyInternetGatewayOutput, error) {
var output ec2.CreateEgressOnlyInternetGatewayOutput
if err := ValidateCreateEgressOnlyInternetGatewayInput(input); err != nil {
return output, err
}
svc := handlers_ec2_eigw.NewNATSEgressOnlyIGWService(natsConn)
result, err := svc.CreateEgressOnlyInternetGateway(input)
if err != nil {
return output, err
}
return *result, nil
}
@@ -0,0 +1,40 @@
package gateway_ec2_eigw
import (
"errors"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/hive/hive/awserrors"
handlers_ec2_eigw "github.com/mulgadc/hive/hive/handlers/ec2/eigw"
"github.com/nats-io/nats.go"
)
// ValidateDeleteEgressOnlyInternetGatewayInput validates the input parameters
func ValidateDeleteEgressOnlyInternetGatewayInput(input *ec2.DeleteEgressOnlyInternetGatewayInput) error {
if input == nil {
return errors.New(awserrors.ErrorInvalidParameterValue)
}
if input.EgressOnlyInternetGatewayId == nil || *input.EgressOnlyInternetGatewayId == "" {
return errors.New(awserrors.ErrorMissingParameter)
}
return nil
}
// DeleteEgressOnlyInternetGateway handles the EC2 DeleteEgressOnlyInternetGateway API call
func DeleteEgressOnlyInternetGateway(input *ec2.DeleteEgressOnlyInternetGatewayInput, natsConn *nats.Conn) (ec2.DeleteEgressOnlyInternetGatewayOutput, error) {
var output ec2.DeleteEgressOnlyInternetGatewayOutput
if err := ValidateDeleteEgressOnlyInternetGatewayInput(input); err != nil {
return output, err
}
svc := handlers_ec2_eigw.NewNATSEgressOnlyIGWService(natsConn)
result, err := svc.DeleteEgressOnlyInternetGateway(input)
if err != nil {
return output, err
}
return *result, nil
}
@@ -0,0 +1,36 @@
package gateway_ec2_eigw
import (
"errors"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/hive/hive/awserrors"
handlers_ec2_eigw "github.com/mulgadc/hive/hive/handlers/ec2/eigw"
"github.com/nats-io/nats.go"
)
// ValidateDescribeEgressOnlyInternetGatewaysInput validates the input parameters
func ValidateDescribeEgressOnlyInternetGatewaysInput(input *ec2.DescribeEgressOnlyInternetGatewaysInput) error {
if input == nil {
return errors.New(awserrors.ErrorInvalidParameterValue)
}
return nil
}
// DescribeEgressOnlyInternetGateways handles the EC2 DescribeEgressOnlyInternetGateways API call
func DescribeEgressOnlyInternetGateways(input *ec2.DescribeEgressOnlyInternetGatewaysInput, natsConn *nats.Conn) (ec2.DescribeEgressOnlyInternetGatewaysOutput, error) {
var output ec2.DescribeEgressOnlyInternetGatewaysOutput
if err := ValidateDescribeEgressOnlyInternetGatewaysInput(input); err != nil {
return output, err
}
svc := handlers_ec2_eigw.NewNATSEgressOnlyIGWService(natsConn)
result, err := svc.DescribeEgressOnlyInternetGateways(input)
if err != nil {
return output, err
}
return *result, nil
}
+127
View File
@@ -0,0 +1,127 @@
package gateway_ec2_eigw
import (
"testing"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/hive/hive/awserrors"
"github.com/stretchr/testify/assert"
)
func TestValidateCreateEgressOnlyInternetGatewayInput(t *testing.T) {
tests := []struct {
name string
input *ec2.CreateEgressOnlyInternetGatewayInput
wantErr string
}{
{
name: "nil input",
input: nil,
wantErr: awserrors.ErrorInvalidParameterValue,
},
{
name: "missing VpcId",
input: &ec2.CreateEgressOnlyInternetGatewayInput{},
wantErr: awserrors.ErrorMissingParameter,
},
{
name: "empty VpcId",
input: &ec2.CreateEgressOnlyInternetGatewayInput{VpcId: aws.String("")},
wantErr: awserrors.ErrorMissingParameter,
},
{
name: "valid input",
input: &ec2.CreateEgressOnlyInternetGatewayInput{VpcId: aws.String("vpc-123")},
wantErr: "",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
err := ValidateCreateEgressOnlyInternetGatewayInput(tt.input)
if tt.wantErr == "" {
assert.NoError(t, err)
} else {
assert.EqualError(t, err, tt.wantErr)
}
})
}
}
func TestValidateDeleteEgressOnlyInternetGatewayInput(t *testing.T) {
tests := []struct {
name string
input *ec2.DeleteEgressOnlyInternetGatewayInput
wantErr string
}{
{
name: "nil input",
input: nil,
wantErr: awserrors.ErrorInvalidParameterValue,
},
{
name: "missing ID",
input: &ec2.DeleteEgressOnlyInternetGatewayInput{},
wantErr: awserrors.ErrorMissingParameter,
},
{
name: "empty ID",
input: &ec2.DeleteEgressOnlyInternetGatewayInput{EgressOnlyInternetGatewayId: aws.String("")},
wantErr: awserrors.ErrorMissingParameter,
},
{
name: "valid input",
input: &ec2.DeleteEgressOnlyInternetGatewayInput{EgressOnlyInternetGatewayId: aws.String("eigw-abc123")},
wantErr: "",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
err := ValidateDeleteEgressOnlyInternetGatewayInput(tt.input)
if tt.wantErr == "" {
assert.NoError(t, err)
} else {
assert.EqualError(t, err, tt.wantErr)
}
})
}
}
func TestValidateDescribeEgressOnlyInternetGatewaysInput(t *testing.T) {
tests := []struct {
name string
input *ec2.DescribeEgressOnlyInternetGatewaysInput
wantErr string
}{
{
name: "nil input",
input: nil,
wantErr: awserrors.ErrorInvalidParameterValue,
},
{
name: "empty input",
input: &ec2.DescribeEgressOnlyInternetGatewaysInput{},
wantErr: "",
},
{
name: "with filter IDs",
input: &ec2.DescribeEgressOnlyInternetGatewaysInput{
EgressOnlyInternetGatewayIds: []*string{aws.String("eigw-abc")},
},
wantErr: "",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
err := ValidateDescribeEgressOnlyInternetGatewaysInput(tt.input)
if tt.wantErr == "" {
assert.NoError(t, err)
} else {
assert.EqualError(t, err, tt.wantErr)
}
})
}
}
+13 -13
View File
@@ -1,13 +1,13 @@
package handlers_ec2_account
import "github.com/aws/aws-sdk-go/service/ec2"
// AccountSettingsService defines the interface for EC2 account-level settings
type AccountSettingsService interface {
EnableEbsEncryptionByDefault(input *ec2.EnableEbsEncryptionByDefaultInput) (*ec2.EnableEbsEncryptionByDefaultOutput, error)
DisableEbsEncryptionByDefault(input *ec2.DisableEbsEncryptionByDefaultInput) (*ec2.DisableEbsEncryptionByDefaultOutput, error)
GetEbsEncryptionByDefault(input *ec2.GetEbsEncryptionByDefaultInput) (*ec2.GetEbsEncryptionByDefaultOutput, error)
GetSerialConsoleAccessStatus(input *ec2.GetSerialConsoleAccessStatusInput) (*ec2.GetSerialConsoleAccessStatusOutput, error)
EnableSerialConsoleAccess(input *ec2.EnableSerialConsoleAccessInput) (*ec2.EnableSerialConsoleAccessOutput, error)
DisableSerialConsoleAccess(input *ec2.DisableSerialConsoleAccessInput) (*ec2.DisableSerialConsoleAccessOutput, error)
}
package handlers_ec2_account
import "github.com/aws/aws-sdk-go/service/ec2"
// AccountSettingsService defines the interface for EC2 account-level settings
type AccountSettingsService interface {
EnableEbsEncryptionByDefault(input *ec2.EnableEbsEncryptionByDefaultInput) (*ec2.EnableEbsEncryptionByDefaultOutput, error)
DisableEbsEncryptionByDefault(input *ec2.DisableEbsEncryptionByDefaultInput) (*ec2.DisableEbsEncryptionByDefaultOutput, error)
GetEbsEncryptionByDefault(input *ec2.GetEbsEncryptionByDefaultInput) (*ec2.GetEbsEncryptionByDefaultOutput, error)
GetSerialConsoleAccessStatus(input *ec2.GetSerialConsoleAccessStatusInput) (*ec2.GetSerialConsoleAccessStatusOutput, error)
EnableSerialConsoleAccess(input *ec2.EnableSerialConsoleAccessInput) (*ec2.EnableSerialConsoleAccessOutput, error)
DisableSerialConsoleAccess(input *ec2.DisableSerialConsoleAccessInput) (*ec2.DisableSerialConsoleAccessOutput, error)
}
+225 -226
View File
@@ -1,226 +1,225 @@
package handlers_ec2_account
import (
"encoding/json"
"fmt"
"log/slog"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/hive/hive/config"
"github.com/nats-io/nats.go"
)
const (
KVBucketAccountSettings = "hive-ec2-account-settings"
KeyEbsEncryptionDefault = "ebs-encryption-default"
KeySerialConsoleAccess = "serial-console-access"
)
// AccountSettingsRecord represents stored account settings
type AccountSettingsRecord struct {
EbsEncryptionByDefault bool `json:"ebs_encryption_by_default"`
SerialConsoleAccess bool `json:"serial_console_access"`
}
// AccountSettingsServiceImpl implements account settings operations with NATS JetStream persistence
type AccountSettingsServiceImpl struct {
config *config.Config
js nats.JetStreamContext
settingsKV nats.KeyValue
}
// NewAccountSettingsServiceImpl creates a new account settings service implementation
func NewAccountSettingsServiceImpl(cfg *config.Config) *AccountSettingsServiceImpl {
return &AccountSettingsServiceImpl{
config: cfg,
}
}
// NewAccountSettingsServiceImplWithNATS creates an account settings service with NATS JetStream for persistence
func NewAccountSettingsServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*AccountSettingsServiceImpl, error) {
js, err := natsConn.JetStream()
if err != nil {
return nil, fmt.Errorf("failed to get JetStream context: %w", err)
}
settingsKV, err := getOrCreateKVBucket(js, KVBucketAccountSettings, 10)
if err != nil {
slog.Warn("Failed to create account settings KV bucket", "error", err)
return NewAccountSettingsServiceImpl(cfg), nil
}
slog.Info("Account settings service initialized with JetStream KV", "bucket", KVBucketAccountSettings)
return &AccountSettingsServiceImpl{
config: cfg,
js: js,
settingsKV: settingsKV,
}, nil
}
func getOrCreateKVBucket(js nats.JetStreamContext, bucketName string, history int) (nats.KeyValue, error) {
kv, err := js.CreateKeyValue(&nats.KeyValueConfig{
Bucket: bucketName,
History: uint8(history),
})
if err != nil {
kv, err = js.KeyValue(bucketName)
if err != nil {
return nil, err
}
}
return kv, nil
}
// getSettings retrieves current account settings
func (s *AccountSettingsServiceImpl) getSettings() (*AccountSettingsRecord, error) {
if s.settingsKV == nil {
// Return defaults if KV not available
return &AccountSettingsRecord{
EbsEncryptionByDefault: false,
SerialConsoleAccess: false,
}, nil
}
entry, err := s.settingsKV.Get("default")
if err != nil {
// Return defaults if not found
return &AccountSettingsRecord{
EbsEncryptionByDefault: false,
SerialConsoleAccess: false,
}, nil
}
var settings AccountSettingsRecord
if err := json.Unmarshal(entry.Value(), &settings); err != nil {
return nil, err
}
return &settings, nil
}
// saveSettings saves current account settings
func (s *AccountSettingsServiceImpl) saveSettings(settings *AccountSettingsRecord) error {
if s.settingsKV == nil {
slog.Warn("Cannot save settings - KV not initialized")
return nil // Return success anyway for in-memory fallback
}
data, err := json.Marshal(settings)
if err != nil {
return err
}
_, err = s.settingsKV.Put("default", data)
return err
}
// EnableEbsEncryptionByDefault enables EBS encryption by default for the account
func (s *AccountSettingsServiceImpl) EnableEbsEncryptionByDefault(input *ec2.EnableEbsEncryptionByDefaultInput) (*ec2.EnableEbsEncryptionByDefaultOutput, error) {
slog.Info("EnableEbsEncryptionByDefault called")
settings, err := s.getSettings()
if err != nil {
return nil, err
}
settings.EbsEncryptionByDefault = true
if err := s.saveSettings(settings); err != nil {
return nil, err
}
return &ec2.EnableEbsEncryptionByDefaultOutput{
EbsEncryptionByDefault: aws.Bool(true),
}, nil
}
// DisableEbsEncryptionByDefault disables EBS encryption by default for the account
func (s *AccountSettingsServiceImpl) DisableEbsEncryptionByDefault(input *ec2.DisableEbsEncryptionByDefaultInput) (*ec2.DisableEbsEncryptionByDefaultOutput, error) {
slog.Info("DisableEbsEncryptionByDefault called")
settings, err := s.getSettings()
if err != nil {
return nil, err
}
settings.EbsEncryptionByDefault = false
if err := s.saveSettings(settings); err != nil {
return nil, err
}
return &ec2.DisableEbsEncryptionByDefaultOutput{
EbsEncryptionByDefault: aws.Bool(false),
}, nil
}
// GetEbsEncryptionByDefault gets the current EBS encryption by default setting
func (s *AccountSettingsServiceImpl) GetEbsEncryptionByDefault(input *ec2.GetEbsEncryptionByDefaultInput) (*ec2.GetEbsEncryptionByDefaultOutput, error) {
slog.Info("GetEbsEncryptionByDefault called")
settings, err := s.getSettings()
if err != nil {
return nil, err
}
return &ec2.GetEbsEncryptionByDefaultOutput{
EbsEncryptionByDefault: aws.Bool(settings.EbsEncryptionByDefault),
}, nil
}
// GetSerialConsoleAccessStatus gets the current serial console access status
func (s *AccountSettingsServiceImpl) GetSerialConsoleAccessStatus(input *ec2.GetSerialConsoleAccessStatusInput) (*ec2.GetSerialConsoleAccessStatusOutput, error) {
slog.Info("GetSerialConsoleAccessStatus called")
settings, err := s.getSettings()
if err != nil {
return nil, err
}
return &ec2.GetSerialConsoleAccessStatusOutput{
SerialConsoleAccessEnabled: aws.Bool(settings.SerialConsoleAccess),
}, nil
}
// EnableSerialConsoleAccess enables serial console access for the account
func (s *AccountSettingsServiceImpl) EnableSerialConsoleAccess(input *ec2.EnableSerialConsoleAccessInput) (*ec2.EnableSerialConsoleAccessOutput, error) {
slog.Info("EnableSerialConsoleAccess called")
settings, err := s.getSettings()
if err != nil {
return nil, err
}
settings.SerialConsoleAccess = true
if err := s.saveSettings(settings); err != nil {
return nil, err
}
return &ec2.EnableSerialConsoleAccessOutput{
SerialConsoleAccessEnabled: aws.Bool(true),
}, nil
}
// DisableSerialConsoleAccess disables serial console access for the account
func (s *AccountSettingsServiceImpl) DisableSerialConsoleAccess(input *ec2.DisableSerialConsoleAccessInput) (*ec2.DisableSerialConsoleAccessOutput, error) {
slog.Info("DisableSerialConsoleAccess called")
settings, err := s.getSettings()
if err != nil {
return nil, err
}
settings.SerialConsoleAccess = false
if err := s.saveSettings(settings); err != nil {
return nil, err
}
return &ec2.DisableSerialConsoleAccessOutput{
SerialConsoleAccessEnabled: aws.Bool(false),
}, nil
}
package handlers_ec2_account
import (
"encoding/json"
"fmt"
"log/slog"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/hive/hive/config"
"github.com/nats-io/nats.go"
)
const (
KVBucketAccountSettings = "hive-ec2-account-settings"
KeyEbsEncryptionDefault = "ebs-encryption-default"
KeySerialConsoleAccess = "serial-console-access"
)
// AccountSettingsRecord represents stored account settings
type AccountSettingsRecord struct {
EbsEncryptionByDefault bool `json:"ebs_encryption_by_default"`
SerialConsoleAccess bool `json:"serial_console_access"`
}
// AccountSettingsServiceImpl implements account settings operations with NATS JetStream persistence
type AccountSettingsServiceImpl struct {
config *config.Config
js nats.JetStreamContext
settingsKV nats.KeyValue
}
// NewAccountSettingsServiceImpl creates a new account settings service implementation
func NewAccountSettingsServiceImpl(cfg *config.Config) *AccountSettingsServiceImpl {
return &AccountSettingsServiceImpl{
config: cfg,
}
}
// NewAccountSettingsServiceImplWithNATS creates an account settings service with NATS JetStream for persistence
func NewAccountSettingsServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*AccountSettingsServiceImpl, error) {
js, err := natsConn.JetStream()
if err != nil {
return nil, fmt.Errorf("failed to get JetStream context: %w", err)
}
settingsKV, err := getOrCreateKVBucket(js, KVBucketAccountSettings, 10)
if err != nil {
slog.Warn("Failed to create account settings KV bucket", "error", err)
return NewAccountSettingsServiceImpl(cfg), nil
}
slog.Info("Account settings service initialized with JetStream KV", "bucket", KVBucketAccountSettings)
return &AccountSettingsServiceImpl{
config: cfg,
js: js,
settingsKV: settingsKV,
}, nil
}
func getOrCreateKVBucket(js nats.JetStreamContext, bucketName string, history int) (nats.KeyValue, error) {
kv, err := js.CreateKeyValue(&nats.KeyValueConfig{
Bucket: bucketName,
History: uint8(history),
})
if err != nil {
kv, err = js.KeyValue(bucketName)
if err != nil {
return nil, err
}
}
return kv, nil
}
// getSettings retrieves current account settings
func (s *AccountSettingsServiceImpl) getSettings() (*AccountSettingsRecord, error) {
if s.settingsKV == nil {
// Return defaults if KV not available
return &AccountSettingsRecord{
EbsEncryptionByDefault: false,
SerialConsoleAccess: false,
}, nil
}
entry, err := s.settingsKV.Get("default")
if err != nil {
// Return defaults if not found
return &AccountSettingsRecord{
EbsEncryptionByDefault: false,
SerialConsoleAccess: false,
}, nil
}
var settings AccountSettingsRecord
if err := json.Unmarshal(entry.Value(), &settings); err != nil {
return nil, err
}
return &settings, nil
}
// saveSettings saves current account settings
func (s *AccountSettingsServiceImpl) saveSettings(settings *AccountSettingsRecord) error {
if s.settingsKV == nil {
slog.Warn("Cannot save settings - KV not initialized")
return nil // Return success anyway for in-memory fallback
}
data, err := json.Marshal(settings)
if err != nil {
return err
}
_, err = s.settingsKV.Put("default", data)
return err
}
// EnableEbsEncryptionByDefault enables EBS encryption by default for the account
func (s *AccountSettingsServiceImpl) EnableEbsEncryptionByDefault(input *ec2.EnableEbsEncryptionByDefaultInput) (*ec2.EnableEbsEncryptionByDefaultOutput, error) {
slog.Info("EnableEbsEncryptionByDefault called")
settings, err := s.getSettings()
if err != nil {
return nil, err
}
settings.EbsEncryptionByDefault = true
if err := s.saveSettings(settings); err != nil {
return nil, err
}
return &ec2.EnableEbsEncryptionByDefaultOutput{
EbsEncryptionByDefault: aws.Bool(true),
}, nil
}
// DisableEbsEncryptionByDefault disables EBS encryption by default for the account
func (s *AccountSettingsServiceImpl) DisableEbsEncryptionByDefault(input *ec2.DisableEbsEncryptionByDefaultInput) (*ec2.DisableEbsEncryptionByDefaultOutput, error) {
slog.Info("DisableEbsEncryptionByDefault called")
settings, err := s.getSettings()
if err != nil {
return nil, err
}
settings.EbsEncryptionByDefault = false
if err := s.saveSettings(settings); err != nil {
return nil, err
}
return &ec2.DisableEbsEncryptionByDefaultOutput{
EbsEncryptionByDefault: aws.Bool(false),
}, nil
}
// GetEbsEncryptionByDefault gets the current EBS encryption by default setting
func (s *AccountSettingsServiceImpl) GetEbsEncryptionByDefault(input *ec2.GetEbsEncryptionByDefaultInput) (*ec2.GetEbsEncryptionByDefaultOutput, error) {
slog.Info("GetEbsEncryptionByDefault called")
settings, err := s.getSettings()
if err != nil {
return nil, err
}
return &ec2.GetEbsEncryptionByDefaultOutput{
EbsEncryptionByDefault: aws.Bool(settings.EbsEncryptionByDefault),
}, nil
}
// GetSerialConsoleAccessStatus gets the current serial console access status
func (s *AccountSettingsServiceImpl) GetSerialConsoleAccessStatus(input *ec2.GetSerialConsoleAccessStatusInput) (*ec2.GetSerialConsoleAccessStatusOutput, error) {
slog.Info("GetSerialConsoleAccessStatus called")
settings, err := s.getSettings()
if err != nil {
return nil, err
}
return &ec2.GetSerialConsoleAccessStatusOutput{
SerialConsoleAccessEnabled: aws.Bool(settings.SerialConsoleAccess),
}, nil
}
// EnableSerialConsoleAccess enables serial console access for the account
func (s *AccountSettingsServiceImpl) EnableSerialConsoleAccess(input *ec2.EnableSerialConsoleAccessInput) (*ec2.EnableSerialConsoleAccessOutput, error) {
slog.Info("EnableSerialConsoleAccess called")
settings, err := s.getSettings()
if err != nil {
return nil, err
}
settings.SerialConsoleAccess = true
if err := s.saveSettings(settings); err != nil {
return nil, err
}
return &ec2.EnableSerialConsoleAccessOutput{
SerialConsoleAccessEnabled: aws.Bool(true),
}, nil
}
// DisableSerialConsoleAccess disables serial console access for the account
func (s *AccountSettingsServiceImpl) DisableSerialConsoleAccess(input *ec2.DisableSerialConsoleAccessInput) (*ec2.DisableSerialConsoleAccessOutput, error) {
slog.Info("DisableSerialConsoleAccess called")
settings, err := s.getSettings()
if err != nil {
return nil, err
}
settings.SerialConsoleAccess = false
if err := s.saveSettings(settings); err != nil {
return nil, err
}
return &ec2.DisableSerialConsoleAccessOutput{
SerialConsoleAccessEnabled: aws.Bool(false),
}, nil
}
@@ -99,4 +99,3 @@ func TestSerialConsole_EnableDisable(t *testing.T) {
require.NoError(t, err)
assert.False(t, *getOut.SerialConsoleAccessEnabled)
}
@@ -41,4 +41,3 @@ func (s *NATSAccountSettingsService) EnableSerialConsoleAccess(input *ec2.Enable
func (s *NATSAccountSettingsService) DisableSerialConsoleAccess(input *ec2.DisableSerialConsoleAccessInput) (*ec2.DisableSerialConsoleAccessOutput, error) {
return utils.NATSRequest[ec2.DisableSerialConsoleAccessOutput](s.natsConn, "ec2.DisableSerialConsoleAccess", input, 30*time.Second)
}
+10
View File
@@ -0,0 +1,10 @@
package handlers_ec2_eigw
import "github.com/aws/aws-sdk-go/service/ec2"
// EgressOnlyIGWService defines the interface for Egress-only Internet Gateway operations
type EgressOnlyIGWService interface {
CreateEgressOnlyInternetGateway(input *ec2.CreateEgressOnlyInternetGatewayInput) (*ec2.CreateEgressOnlyInternetGatewayOutput, error)
DeleteEgressOnlyInternetGateway(input *ec2.DeleteEgressOnlyInternetGatewayInput) (*ec2.DeleteEgressOnlyInternetGatewayOutput, error)
DescribeEgressOnlyInternetGateways(input *ec2.DescribeEgressOnlyInternetGatewaysInput) (*ec2.DescribeEgressOnlyInternetGatewaysOutput, error)
}
+217
View File
@@ -0,0 +1,217 @@
package handlers_ec2_eigw
import (
"crypto/rand"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"log/slog"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/hive/hive/awserrors"
"github.com/mulgadc/hive/hive/config"
"github.com/mulgadc/hive/hive/utils"
"github.com/nats-io/nats.go"
)
// Ensure EgressOnlyIGWServiceImpl implements EgressOnlyIGWService
var _ EgressOnlyIGWService = (*EgressOnlyIGWServiceImpl)(nil)
const KVBucketEgressOnlyIGW = "hive-eigw"
// EgressOnlyIGWRecord represents a stored Egress-only Internet Gateway
type EgressOnlyIGWRecord struct {
EgressOnlyInternetGatewayId string `json:"egress_only_internet_gateway_id"`
VpcId string `json:"vpc_id"`
State string `json:"state"`
Tags map[string]string `json:"tags"`
CreatedAt time.Time `json:"created_at"`
}
// EgressOnlyIGWServiceImpl implements Egress-only Internet Gateway operations with NATS JetStream persistence
type EgressOnlyIGWServiceImpl struct {
config *config.Config
eigwKV 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()
if err != nil {
return nil, fmt.Errorf("failed to get JetStream context: %w", err)
}
eigwKV, err := getOrCreateKVBucket(js, KVBucketEgressOnlyIGW, 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketEgressOnlyIGW, err)
}
slog.Info("Egress-only IGW service initialized with JetStream KV", "bucket", KVBucketEgressOnlyIGW)
return &EgressOnlyIGWServiceImpl{
config: cfg,
eigwKV: eigwKV,
}, 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
}
// generateEgressOnlyIGWID generates a unique Egress-only Internet Gateway ID
func generateEgressOnlyIGWID() string {
b := make([]byte, 8)
if _, err := rand.Read(b); err != nil {
panic("crypto/rand failed: " + err.Error())
}
return "eigw-" + hex.EncodeToString(b)
}
// CreateEgressOnlyInternetGateway creates a new Egress-only Internet Gateway
func (s *EgressOnlyIGWServiceImpl) CreateEgressOnlyInternetGateway(input *ec2.CreateEgressOnlyInternetGatewayInput) (*ec2.CreateEgressOnlyInternetGatewayOutput, error) {
if input.VpcId == nil || *input.VpcId == "" {
return nil, errors.New(awserrors.ErrorMissingParameter)
}
eigwID := generateEgressOnlyIGWID()
record := EgressOnlyIGWRecord{
EgressOnlyInternetGatewayId: eigwID,
VpcId: *input.VpcId,
State: "attached",
Tags: make(map[string]string),
CreatedAt: time.Now(),
}
for _, tagSpec := range input.TagSpecifications {
if tagSpec.ResourceType != nil && *tagSpec.ResourceType == "egress-only-internet-gateway" {
for _, tag := range tagSpec.Tags {
if tag.Key != nil && tag.Value != nil {
record.Tags[*tag.Key] = *tag.Value
}
}
}
}
data, err := json.Marshal(record)
if err != nil {
return nil, fmt.Errorf("failed to marshal Egress-only IGW record: %w", err)
}
if _, err := s.eigwKV.Put(eigwID, data); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
slog.Info("CreateEgressOnlyInternetGateway completed", "egressOnlyInternetGatewayId", eigwID, "vpcId", record.VpcId)
return &ec2.CreateEgressOnlyInternetGatewayOutput{
EgressOnlyInternetGateway: s.recordToEC2(&record),
}, nil
}
// DeleteEgressOnlyInternetGateway deletes an Egress-only Internet Gateway
func (s *EgressOnlyIGWServiceImpl) DeleteEgressOnlyInternetGateway(input *ec2.DeleteEgressOnlyInternetGatewayInput) (*ec2.DeleteEgressOnlyInternetGatewayOutput, error) {
if input.EgressOnlyInternetGatewayId == nil || *input.EgressOnlyInternetGatewayId == "" {
return nil, errors.New(awserrors.ErrorMissingParameter)
}
eigwID := *input.EgressOnlyInternetGatewayId
if _, err := s.eigwKV.Get(eigwID); err != nil {
return nil, errors.New(awserrors.ErrorInvalidEgressOnlyInternetGatewayIdNotFound)
}
if err := s.eigwKV.Delete(eigwID); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
slog.Info("DeleteEgressOnlyInternetGateway completed", "egressOnlyInternetGatewayId", eigwID)
return &ec2.DeleteEgressOnlyInternetGatewayOutput{
ReturnCode: aws.Bool(true),
}, nil
}
// DescribeEgressOnlyInternetGateways describes Egress-only Internet Gateways
func (s *EgressOnlyIGWServiceImpl) DescribeEgressOnlyInternetGateways(input *ec2.DescribeEgressOnlyInternetGatewaysInput) (*ec2.DescribeEgressOnlyInternetGatewaysOutput, error) {
var egressOnlyIGWs []*ec2.EgressOnlyInternetGateway
eigwIDs := make(map[string]bool)
for _, id := range input.EgressOnlyInternetGatewayIds {
if id != nil {
eigwIDs[*id] = true
}
}
keys, err := s.eigwKV.Keys()
if err != nil && !errors.Is(err, nats.ErrNoKeysFound) {
return nil, errors.New(awserrors.ErrorServerInternal)
}
for _, key := range keys {
if len(eigwIDs) > 0 && !eigwIDs[key] {
continue
}
entry, err := s.eigwKV.Get(key)
if err != nil {
slog.Warn("Failed to get Egress-only IGW record", "key", key, "error", err)
continue
}
var record EgressOnlyIGWRecord
if err := json.Unmarshal(entry.Value(), &record); err != nil {
slog.Warn("Failed to unmarshal Egress-only IGW record", "key", key, "error", err)
continue
}
egressOnlyIGWs = append(egressOnlyIGWs, s.recordToEC2(&record))
}
slog.Info("DescribeEgressOnlyInternetGateways completed", "count", len(egressOnlyIGWs))
return &ec2.DescribeEgressOnlyInternetGatewaysOutput{
EgressOnlyInternetGateways: egressOnlyIGWs,
}, nil
}
func (s *EgressOnlyIGWServiceImpl) recordToEC2(record *EgressOnlyIGWRecord) *ec2.EgressOnlyInternetGateway {
eigw := &ec2.EgressOnlyInternetGateway{
EgressOnlyInternetGatewayId: aws.String(record.EgressOnlyInternetGatewayId),
Attachments: []*ec2.InternetGatewayAttachment{
{
VpcId: aws.String(record.VpcId),
State: aws.String(record.State),
},
},
}
if len(record.Tags) > 0 {
tags := make([]*ec2.Tag, 0, len(record.Tags))
for k, v := range record.Tags {
tags = append(tags, &ec2.Tag{Key: aws.String(k), Value: aws.String(v)})
}
eigw.Tags = tags
}
return eigw
}
+186
View File
@@ -0,0 +1,186 @@
package handlers_ec2_eigw
import (
"testing"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/nats-io/nats-server/v2/server"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func setupTestEIGWService(t *testing.T) *EgressOnlyIGWServiceImpl {
t.Helper()
opts := &server.Options{
Host: "127.0.0.1",
Port: -1,
JetStream: true,
StoreDir: t.TempDir(),
NoLog: true,
NoSigs: true,
}
ns, err := server.NewServer(opts)
require.NoError(t, err)
go ns.Start()
require.True(t, ns.ReadyForConnections(5*time.Second))
t.Cleanup(func() { ns.Shutdown() })
nc, err := nats.Connect(ns.ClientURL())
require.NoError(t, err)
t.Cleanup(func() { nc.Close() })
svc, err := NewEgressOnlyIGWServiceImplWithNATS(nil, nc)
require.NoError(t, err)
return svc
}
func createTestEIGW(t *testing.T, svc *EgressOnlyIGWServiceImpl) string {
t.Helper()
out, err := svc.CreateEgressOnlyInternetGateway(&ec2.CreateEgressOnlyInternetGatewayInput{
VpcId: aws.String("vpc-test123"),
})
require.NoError(t, err)
return *out.EgressOnlyInternetGateway.EgressOnlyInternetGatewayId
}
func TestCreateEgressOnlyInternetGateway(t *testing.T) {
svc := setupTestEIGWService(t)
out, err := svc.CreateEgressOnlyInternetGateway(&ec2.CreateEgressOnlyInternetGatewayInput{
VpcId: aws.String("vpc-test123"),
})
require.NoError(t, err)
require.NotNil(t, out.EgressOnlyInternetGateway)
assert.Equal(t, "eigw-", (*out.EgressOnlyInternetGateway.EgressOnlyInternetGatewayId)[:5])
require.NotEmpty(t, out.EgressOnlyInternetGateway.Attachments)
assert.Equal(t, "vpc-test123", *out.EgressOnlyInternetGateway.Attachments[0].VpcId)
assert.Equal(t, "attached", *out.EgressOnlyInternetGateway.Attachments[0].State)
}
func TestCreateEgressOnlyInternetGateway_MissingVpcId(t *testing.T) {
svc := setupTestEIGWService(t)
_, err := svc.CreateEgressOnlyInternetGateway(&ec2.CreateEgressOnlyInternetGatewayInput{})
assert.ErrorContains(t, err, "MissingParameter")
}
func TestCreateEgressOnlyInternetGateway_EmptyVpcId(t *testing.T) {
svc := setupTestEIGWService(t)
_, err := svc.CreateEgressOnlyInternetGateway(&ec2.CreateEgressOnlyInternetGatewayInput{
VpcId: aws.String(""),
})
assert.ErrorContains(t, err, "MissingParameter")
}
func TestCreateEgressOnlyInternetGateway_WithTags(t *testing.T) {
svc := setupTestEIGWService(t)
out, err := svc.CreateEgressOnlyInternetGateway(&ec2.CreateEgressOnlyInternetGatewayInput{
VpcId: aws.String("vpc-tagged"),
TagSpecifications: []*ec2.TagSpecification{
{
ResourceType: aws.String("egress-only-internet-gateway"),
Tags: []*ec2.Tag{
{Key: aws.String("Name"), Value: aws.String("my-eigw")},
{Key: aws.String("Env"), Value: aws.String("test")},
},
},
},
})
require.NoError(t, err)
require.NotNil(t, out.EgressOnlyInternetGateway)
assert.Len(t, out.EgressOnlyInternetGateway.Tags, 2)
// Verify tags persist through describe
desc, err := svc.DescribeEgressOnlyInternetGateways(&ec2.DescribeEgressOnlyInternetGatewaysInput{
EgressOnlyInternetGatewayIds: []*string{out.EgressOnlyInternetGateway.EgressOnlyInternetGatewayId},
})
require.NoError(t, err)
require.Len(t, desc.EgressOnlyInternetGateways, 1)
assert.Len(t, desc.EgressOnlyInternetGateways[0].Tags, 2)
}
func TestCreateEgressOnlyInternetGateway_TagsWrongResourceType(t *testing.T) {
svc := setupTestEIGWService(t)
out, err := svc.CreateEgressOnlyInternetGateway(&ec2.CreateEgressOnlyInternetGatewayInput{
VpcId: aws.String("vpc-tagged2"),
TagSpecifications: []*ec2.TagSpecification{
{
ResourceType: aws.String("instance"),
Tags: []*ec2.Tag{
{Key: aws.String("Name"), Value: aws.String("wrong-type")},
},
},
},
})
require.NoError(t, err)
assert.Empty(t, out.EgressOnlyInternetGateway.Tags)
}
func TestDeleteEgressOnlyInternetGateway(t *testing.T) {
svc := setupTestEIGWService(t)
eigwID := createTestEIGW(t, svc)
out, err := svc.DeleteEgressOnlyInternetGateway(&ec2.DeleteEgressOnlyInternetGatewayInput{
EgressOnlyInternetGatewayId: aws.String(eigwID),
})
require.NoError(t, err)
assert.True(t, *out.ReturnCode)
desc, err := svc.DescribeEgressOnlyInternetGateways(&ec2.DescribeEgressOnlyInternetGatewaysInput{
EgressOnlyInternetGatewayIds: []*string{aws.String(eigwID)},
})
require.NoError(t, err)
assert.Empty(t, desc.EgressOnlyInternetGateways)
}
func TestDeleteEgressOnlyInternetGateway_NotFound(t *testing.T) {
svc := setupTestEIGWService(t)
_, err := svc.DeleteEgressOnlyInternetGateway(&ec2.DeleteEgressOnlyInternetGatewayInput{
EgressOnlyInternetGatewayId: aws.String("eigw-nonexistent"),
})
assert.ErrorContains(t, err, "InvalidEgressOnlyInternetGatewayId.NotFound")
}
func TestDeleteEgressOnlyInternetGateway_MissingID(t *testing.T) {
svc := setupTestEIGWService(t)
_, err := svc.DeleteEgressOnlyInternetGateway(&ec2.DeleteEgressOnlyInternetGatewayInput{})
assert.ErrorContains(t, err, "MissingParameter")
}
func TestDeleteEgressOnlyInternetGateway_EmptyID(t *testing.T) {
svc := setupTestEIGWService(t)
_, err := svc.DeleteEgressOnlyInternetGateway(&ec2.DeleteEgressOnlyInternetGatewayInput{
EgressOnlyInternetGatewayId: aws.String(""),
})
assert.ErrorContains(t, err, "MissingParameter")
}
func TestDescribeEgressOnlyInternetGateways_All(t *testing.T) {
svc := setupTestEIGWService(t)
createTestEIGW(t, svc)
createTestEIGW(t, svc)
desc, err := svc.DescribeEgressOnlyInternetGateways(&ec2.DescribeEgressOnlyInternetGatewaysInput{})
require.NoError(t, err)
assert.Len(t, desc.EgressOnlyInternetGateways, 2)
}
func TestDescribeEgressOnlyInternetGateways_ByID(t *testing.T) {
svc := setupTestEIGWService(t)
eigwID := createTestEIGW(t, svc)
desc, err := svc.DescribeEgressOnlyInternetGateways(&ec2.DescribeEgressOnlyInternetGatewaysInput{
EgressOnlyInternetGatewayIds: []*string{aws.String(eigwID)},
})
require.NoError(t, err)
require.Len(t, desc.EgressOnlyInternetGateways, 1)
assert.Equal(t, eigwID, *desc.EgressOnlyInternetGateways[0].EgressOnlyInternetGatewayId)
}
func TestDescribeEgressOnlyInternetGateways_Empty(t *testing.T) {
svc := setupTestEIGWService(t)
desc, err := svc.DescribeEgressOnlyInternetGateways(&ec2.DescribeEgressOnlyInternetGatewaysInput{})
require.NoError(t, err)
assert.Empty(t, desc.EgressOnlyInternetGateways)
}
+31
View File
@@ -0,0 +1,31 @@
package handlers_ec2_eigw
import (
"time"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/hive/hive/utils"
"github.com/nats-io/nats.go"
)
// NATSEgressOnlyIGWService handles Egress-only Internet Gateway operations via NATS messaging
type NATSEgressOnlyIGWService struct {
natsConn *nats.Conn
}
// NewNATSEgressOnlyIGWService creates a new NATS-based Egress-only Internet Gateway service
func NewNATSEgressOnlyIGWService(conn *nats.Conn) EgressOnlyIGWService {
return &NATSEgressOnlyIGWService{natsConn: conn}
}
func (s *NATSEgressOnlyIGWService) CreateEgressOnlyInternetGateway(input *ec2.CreateEgressOnlyInternetGatewayInput) (*ec2.CreateEgressOnlyInternetGatewayOutput, error) {
return utils.NATSRequest[ec2.CreateEgressOnlyInternetGatewayOutput](s.natsConn, "ec2.CreateEgressOnlyInternetGateway", input, 30*time.Second)
}
func (s *NATSEgressOnlyIGWService) DeleteEgressOnlyInternetGateway(input *ec2.DeleteEgressOnlyInternetGatewayInput) (*ec2.DeleteEgressOnlyInternetGatewayOutput, error) {
return utils.NATSRequest[ec2.DeleteEgressOnlyInternetGatewayOutput](s.natsConn, "ec2.DeleteEgressOnlyInternetGateway", input, 30*time.Second)
}
func (s *NATSEgressOnlyIGWService) DescribeEgressOnlyInternetGateways(input *ec2.DescribeEgressOnlyInternetGatewaysInput) (*ec2.DescribeEgressOnlyInternetGatewaysOutput, error) {
return utils.NATSRequest[ec2.DescribeEgressOnlyInternetGatewaysOutput](s.natsConn, "ec2.DescribeEgressOnlyInternetGateways", input, 30*time.Second)
}
+3
View File
@@ -78,6 +78,9 @@ func getResourceType(resourceID string) string {
if strings.HasPrefix(resourceID, "igw-") {
return "internet-gateway"
}
if strings.HasPrefix(resourceID, "eigw-") {
return "egress-only-internet-gateway"
}
return "unknown"
}
@@ -436,6 +436,7 @@ func TestGetResourceType(t *testing.T) {
{"sg-abc123", "security-group"},
{"rtb-abc123", "route-table"},
{"igw-abc123", "internet-gateway"},
{"eigw-abc123", "egress-only-internet-gateway"},
{"unknown-abc123", "unknown"},
}
+11
View File
@@ -47,6 +47,17 @@ func SafeInt64ToUint64(v int64) uint64 {
return uint64(v)
}
// SafeIntToUint8 converts int to uint8, clamping to [0, 255]
func SafeIntToUint8(v int) uint8 {
if v < 0 {
return 0
}
if v > math.MaxUint8 {
return math.MaxUint8
}
return uint8(v)
}
// SafeIntToUint64 converts int to uint64, returning 0 if negative
func SafeIntToUint64(v int) uint64 {
if v < 0 {