From 096b3380fc86e20b9186d7648f3ddf91dc96c378 Mon Sep 17 00:00:00 2001 From: Julian Sommer <115516918+juliansommer@users.noreply.github.com> Date: Fri, 13 Mar 2026 12:18:58 +1000 Subject: [PATCH] feat: default vpc account scoping (#74) --- cmd/hive/cmd/admin.go | 28 +++++-- hive/daemon/daemon.go | 2 + hive/daemon/daemon_handlers_vpc.go | 22 ++++++ hive/handlers/ec2/vpc/service_impl.go | 26 ------- hive/handlers/ec2/vpc/service_impl_test.go | 91 ++++++++++++++++++++++ hive/handlers/iam/account_test.go | 57 ++++++++++++++ hive/handlers/iam/service_impl.go | 15 ++++ 7 files changed, 207 insertions(+), 34 deletions(-) diff --git a/cmd/hive/cmd/admin.go b/cmd/hive/cmd/admin.go index f7f35fd18..c534e79e8 100644 --- a/cmd/hive/cmd/admin.go +++ b/cmd/hive/cmd/admin.go @@ -37,11 +37,13 @@ import ( "github.com/mulgadc/hive/hive/admin" "github.com/mulgadc/hive/hive/config" "github.com/mulgadc/hive/hive/formation" + handlers_ec2_vpc "github.com/mulgadc/hive/hive/handlers/ec2/vpc" handlers_iam "github.com/mulgadc/hive/hive/handlers/iam" "github.com/mulgadc/hive/hive/utils" "github.com/mulgadc/viperblock/viperblock" "github.com/mulgadc/viperblock/viperblock/backends/s3" "github.com/mulgadc/viperblock/viperblock/v_utils" + "github.com/nats-io/nats.go" "github.com/pterm/pterm" "github.com/spf13/cobra" ) @@ -1325,26 +1327,26 @@ func buildRemoteNodes(allNodes map[string]formation.NodeInfo, localNode string) // initIAMServiceFromConfig loads config, connects to NATS, loads the master // key, and returns an initialised IAMServiceImpl. Callers must defer nc.Close(). -func initIAMServiceFromConfig() (*handlers_iam.IAMServiceImpl, *config.ClusterConfig, func(), error) { +func initIAMServiceFromConfig() (*handlers_iam.IAMServiceImpl, *config.ClusterConfig, *nats.Conn, func(), error) { cfg, nc, err := loadConfigAndConnect() if err != nil { - return nil, nil, nil, fmt.Errorf("connect to cluster: %w", err) + return nil, nil, nil, nil, fmt.Errorf("connect to cluster: %w", err) } masterKeyPath := filepath.Join(cfg.NodeBaseDir(), "config", "master.key") masterKey, err := handlers_iam.LoadMasterKey(masterKeyPath) if err != nil { nc.Close() - return nil, nil, nil, fmt.Errorf("load master key: %w", err) + return nil, nil, nil, nil, fmt.Errorf("load master key: %w", err) } svc, err := handlers_iam.NewIAMServiceImpl(nc, masterKey) if err != nil { nc.Close() - return nil, nil, nil, fmt.Errorf("init IAM service: %w", err) + return nil, nil, nil, nil, fmt.Errorf("init IAM service: %w", err) } - return svc, cfg, func() { nc.Close() }, nil + return svc, cfg, nc, func() { nc.Close() }, nil } // adminAccessPolicyDocument is the AdministratorAccess policy document that @@ -1354,7 +1356,7 @@ const adminAccessPolicyDocument = `{"Version":"2012-10-17","Statement":[{"Effect func runAccountCreate(cmd *cobra.Command, args []string) { name, _ := cmd.Flags().GetString("name") - svc, cfg, cleanup, err := initIAMServiceFromConfig() + svc, cfg, nc, cleanup, err := initIAMServiceFromConfig() if err != nil { fmt.Fprintf(os.Stderr, "Error: %v\n", err) os.Exit(1) @@ -1369,6 +1371,16 @@ func runAccountCreate(cmd *cobra.Command, args []string) { } accountID := account.AccountID + // Create default VPC for the new account (belt-and-suspenders: daemon also + // does this via iam.account.created event, but daemon may not be running). + nodeConfig := cfg.Nodes[cfg.Node] + vpcSvc, vpcErr := handlers_ec2_vpc.NewVPCServiceImplWithNATS(&nodeConfig, nc) + if vpcErr != nil { + fmt.Fprintf(os.Stderr, "Warning: could not create default VPC service: %v\n", vpcErr) + } else if vpcErr = vpcSvc.EnsureDefaultVPC(accountID); vpcErr != nil { + fmt.Fprintf(os.Stderr, "Warning: could not create default VPC: %v\n", vpcErr) + } + // 2. Create admin user _, err = svc.CreateUser(accountID, &iam.CreateUserInput{ UserName: aws.String("admin"), @@ -1414,7 +1426,7 @@ func runAccountCreate(cmd *cobra.Command, args []string) { endpointHost := "localhost" certPath := filepath.Join(cfg.NodeBaseDir(), "config", "ca.pem") - nodeConfig := cfg.Nodes[cfg.Node] + nodeConfig = cfg.Nodes[cfg.Node] if h, _, err := net.SplitHostPort(nodeConfig.AWSGW.Host); err == nil { if h != "" && h != "0.0.0.0" { endpointHost = h @@ -1457,7 +1469,7 @@ func runAccountCreate(cmd *cobra.Command, args []string) { } func runAccountList(cmd *cobra.Command, args []string) { - svc, _, cleanup, err := initIAMServiceFromConfig() + svc, _, _, cleanup, err := initIAMServiceFromConfig() if err != nil { fmt.Fprintf(os.Stderr, "Error: %v\n", err) os.Exit(1) diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index d41ddc4c3..db629d836 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -403,6 +403,8 @@ func (d *Daemon) subscribeAll() error { {"hive.nodes.discover", d.handleNodeDiscover, ""}, {"hive.node.status", d.handleNodeStatus, ""}, {"hive.node.vms", d.handleNodeVMs, ""}, + // Account creation → create default VPC for new account + {"iam.account.created", d.handleAccountCreated, "hive-workers"}, // Coordinated cluster shutdown phases (fan-out, no queue group) {"hive.cluster.shutdown.gate", d.handleShutdownGate, ""}, {"hive.cluster.shutdown.drain", d.handleShutdownDrain, ""}, diff --git a/hive/daemon/daemon_handlers_vpc.go b/hive/daemon/daemon_handlers_vpc.go index fe9647118..b1ccd219e 100644 --- a/hive/daemon/daemon_handlers_vpc.go +++ b/hive/daemon/daemon_handlers_vpc.go @@ -1,6 +1,9 @@ package daemon import ( + "encoding/json" + "log/slog" + "github.com/nats-io/nats.go" ) @@ -39,3 +42,22 @@ func (d *Daemon) handleEC2DeleteNetworkInterface(msg *nats.Msg) { func (d *Daemon) handleEC2DescribeNetworkInterfaces(msg *nats.Msg) { handleNATSRequest(msg, d.vpcService.DescribeNetworkInterfaces) } + +// handleAccountCreated creates a default VPC for a newly created account. +func (d *Daemon) handleAccountCreated(msg *nats.Msg) { + var evt struct { + AccountID string `json:"account_id"` + } + if err := json.Unmarshal(msg.Data, &evt); err != nil { + slog.Error("Failed to unmarshal account creation event", "error", err) + return + } + if evt.AccountID == "" { + slog.Error("Account creation event has empty account ID") + return + } + if err := d.vpcService.EnsureDefaultVPC(evt.AccountID); err != nil { + slog.Error("Failed to create default VPC for new account", + "accountID", evt.AccountID, "error", err) + } +} diff --git a/hive/handlers/ec2/vpc/service_impl.go b/hive/handlers/ec2/vpc/service_impl.go index b803e5014..d1dfecfaa 100644 --- a/hive/handlers/ec2/vpc/service_impl.go +++ b/hive/handlers/ec2/vpc/service_impl.go @@ -294,32 +294,6 @@ func (s *VPCServiceImpl) DescribeVpcs(input *ec2.DescribeVpcsInput, accountID st vpcs = append(vpcs, s.vpcRecordToEC2(&record, accountID)) } - // Include global default VPCs (created at startup) for non-root accounts. - // This matches AWS behavior where all accounts have a default VPC. - if accountID != utils.GlobalAccountID { - globalPrefix := utils.GlobalAccountID + "." - for _, key := range keys { - if !strings.HasPrefix(key, globalPrefix) { - continue - } - entry, err := s.vpcKV.Get(key) - if err != nil { - continue - } - var record VPCRecord - if err := json.Unmarshal(entry.Value(), &record); err != nil { - continue - } - if !record.IsDefault { - continue - } - if len(vpcIDs) > 0 && !vpcIDs[record.VpcId] { - continue - } - vpcs = append(vpcs, s.vpcRecordToEC2(&record, accountID)) - } - } - // If specific VPC IDs were requested but not found, return error if len(vpcIDs) > 0 { found := make(map[string]bool) diff --git a/hive/handlers/ec2/vpc/service_impl_test.go b/hive/handlers/ec2/vpc/service_impl_test.go index c4588acd4..23b914d14 100644 --- a/hive/handlers/ec2/vpc/service_impl_test.go +++ b/hive/handlers/ec2/vpc/service_impl_test.go @@ -1,6 +1,7 @@ package handlers_ec2_vpc import ( + "encoding/json" "testing" "time" @@ -873,6 +874,96 @@ func TestDetachENI_NotFound(t *testing.T) { assert.ErrorContains(t, err, "InvalidNetworkInterfaceID.NotFound") } +// --- Per-account isolation tests --- + +func TestEnsureDefaultVPC_PerAccountIsolation(t *testing.T) { + svc := setupTestVPCService(t) + accountA := "111111111111" + accountB := "222222222222" + + require.NoError(t, svc.EnsureDefaultVPC(accountA)) + require.NoError(t, svc.EnsureDefaultVPC(accountB)) + + // Each account should see only their own default VPC + descA, err := svc.DescribeVpcs(&ec2.DescribeVpcsInput{}, accountA) + require.NoError(t, err) + require.Len(t, descA.Vpcs, 1) + assert.Equal(t, accountA, *descA.Vpcs[0].OwnerId) + + descB, err := svc.DescribeVpcs(&ec2.DescribeVpcsInput{}, accountB) + require.NoError(t, err) + require.Len(t, descB.Vpcs, 1) + assert.Equal(t, accountB, *descB.Vpcs[0].OwnerId) + + // VPC IDs should be different + assert.NotEqual(t, *descA.Vpcs[0].VpcId, *descB.Vpcs[0].VpcId) +} + +func TestEnsureDefaultVPC_IndependentVNIs(t *testing.T) { + svc, nc := setupTestVPCServiceWithNC(t) + accountA := "111111111111" + accountB := "222222222222" + + // Capture VNIs from vpc.create events + vniCh := make(chan int64, 2) + sub, err := nc.Subscribe("vpc.create", func(msg *nats.Msg) { + var evt struct { + VNI int64 `json:"vni"` + } + if err := json.Unmarshal(msg.Data, &evt); err == nil { + vniCh <- evt.VNI + } + }) + require.NoError(t, err) + defer func() { _ = sub.Unsubscribe() }() + + require.NoError(t, svc.EnsureDefaultVPC(accountA)) + require.NoError(t, svc.EnsureDefaultVPC(accountB)) + + var vnis []int64 + for range 2 { + select { + case vni := <-vniCh: + vnis = append(vnis, vni) + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for vpc.create event") + } + } + assert.NotEqual(t, vnis[0], vnis[1], "each account should get a unique VNI") +} + +func TestDescribeVpcs_NoGlobalSharing(t *testing.T) { + svc := setupTestVPCService(t) + globalAccount := "000000000000" + otherAccount := "111111111111" + + // Create default VPC for global account only + require.NoError(t, svc.EnsureDefaultVPC(globalAccount)) + + // Other account should NOT see the global default VPC + desc, err := svc.DescribeVpcs(&ec2.DescribeVpcsInput{}, otherAccount) + require.NoError(t, err) + assert.Empty(t, desc.Vpcs) +} + +func TestGetDefaultSubnet_PerAccount(t *testing.T) { + svc := setupTestVPCService(t) + accountA := "111111111111" + accountB := "222222222222" + + require.NoError(t, svc.EnsureDefaultVPC(accountA)) + require.NoError(t, svc.EnsureDefaultVPC(accountB)) + + subA, err := svc.GetDefaultSubnet(accountA) + require.NoError(t, err) + subB, err := svc.GetDefaultSubnet(accountB) + require.NoError(t, err) + + assert.NotEqual(t, subA.SubnetId, subB.SubnetId) + assert.Equal(t, "172.31.0.0/20", subA.CidrBlock) + assert.Equal(t, "172.31.0.0/20", subB.CidrBlock) +} + // --- EnsureDefaultVPC event test --- func TestEnsureDefaultVPC_PublishesEvents(t *testing.T) { diff --git a/hive/handlers/iam/account_test.go b/hive/handlers/iam/account_test.go index a576a8eee..e46a228eb 100644 --- a/hive/handlers/iam/account_test.go +++ b/hive/handlers/iam/account_test.go @@ -1,11 +1,15 @@ package handlers_iam import ( + "encoding/json" "testing" + "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/iam" "github.com/mulgadc/hive/hive/utils" + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -377,3 +381,56 @@ func TestSeedBootstrap_AccountScoped(t *testing.T) { assert.Equal(t, "system", account.AccountName) assert.Equal(t, "ACTIVE", account.Status) } + +// ============================================================================ +// Account Creation Event Tests +// ============================================================================ + +func TestCreateAccount_PublishesEvent(t *testing.T) { + 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() }) + + masterKey, err := GenerateMasterKey() + require.NoError(t, err) + + svc, err := NewIAMServiceImpl(nc, masterKey) + require.NoError(t, err) + + eventCh := make(chan *nats.Msg, 1) + sub, err := nc.Subscribe("iam.account.created", func(msg *nats.Msg) { + eventCh <- msg + }) + require.NoError(t, err) + defer func() { _ = sub.Unsubscribe() }() + + acc, err := svc.CreateAccount("Event Test") + require.NoError(t, err) + + select { + case msg := <-eventCh: + var evt struct { + AccountID string `json:"account_id"` + AccountName string `json:"account_name"` + } + require.NoError(t, json.Unmarshal(msg.Data, &evt)) + assert.Equal(t, acc.AccountID, evt.AccountID) + assert.Equal(t, "Event Test", evt.AccountName) + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for iam.account.created event") + } +} diff --git a/hive/handlers/iam/service_impl.go b/hive/handlers/iam/service_impl.go index c333891d5..bea40af03 100644 --- a/hive/handlers/iam/service_impl.go +++ b/hive/handlers/iam/service_impl.go @@ -33,6 +33,7 @@ const ( // IAMServiceImpl implements IAM operations using NATS JetStream KV. type IAMServiceImpl struct { js nats.JetStreamContext + natsConn *nats.Conn usersBucket nats.KeyValue accessKeysBucket nats.KeyValue policiesBucket nats.KeyValue @@ -91,6 +92,7 @@ func NewIAMServiceImpl(natsConn *nats.Conn, masterKey []byte) (*IAMServiceImpl, return &IAMServiceImpl{ js: js, + natsConn: natsConn, usersBucket: usersBucket, accessKeysBucket: accessKeysBucket, policiesBucket: policiesBucket, @@ -772,6 +774,19 @@ func (s *IAMServiceImpl) CreateAccount(name string) (*Account, error) { } slog.Info("Account created", "accountID", accountID, "name", name) + + if s.natsConn != nil { + evt, err := json.Marshal(struct { + AccountID string `json:"account_id"` + AccountName string `json:"account_name"` + }{AccountID: accountID, AccountName: name}) + if err != nil { + slog.Error("Failed to marshal account creation event", "accountID", accountID, "error", err) + } else if err := s.natsConn.Publish("iam.account.created", evt); err != nil { + slog.Error("Failed to publish account creation event", "accountID", accountID, "error", err) + } + } + return &account, nil }