gsadmin/spinifex
1
0
Fork 0
mirror of https://github.com/mulgadc/spinifex.git synced 2026-10-21 19:56:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: default vpc account scoping (#74)

Browse Source
This commit is contained in:
Julian Sommer
2026-03-13 12:18:58 +10:00
committed by GitHub
parent 663805cd23
commit 096b3380fc
7 changed files with 207 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/hive/cmd/admin.go
+20 -8
View File
@@ -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)
hive/daemon/daemon.go
+2
View File
@@ -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, ""},
hive/daemon/daemon_handlers_vpc.go
+22
View File
@@ -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)
}
}
hive/handlers/ec2/vpc/service_impl.go
-26
View File
@@ -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)
hive/handlers/ec2/vpc/service_impl_test.go
+91
View File
@@ -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) {
hive/handlers/iam/account_test.go
+57
View File
@@ -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")
}
}
hive/handlers/iam/service_impl.go
+15
View File
@@ -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
}