diff --git a/docs/service-interfaces.yaml b/docs/service-interfaces.yaml index 9c6f4b287..48062e8d8 100644 --- a/docs/service-interfaces.yaml +++ b/docs/service-interfaces.yaml @@ -139,6 +139,8 @@ services: - "ebs.delete" - "ebs.sync" - "ebs.snapshot.*" + - "ebs.config" + - "ebs.config.*" # → predastore (S3 / AMI / keypair store) - "s3.*" # IAM lifecycle event @@ -187,6 +189,8 @@ services: - "ebs.delete" - "ebs.sync" - "ebs.snapshot.*" + - "ebs.config" + - "ebs.config.*" publishes: - "ebs.mount.response" - "ebs.unmount.response" diff --git a/go.mod b/go.mod index 31588aa37..18e1330e7 100644 --- a/go.mod +++ b/go.mod @@ -14,7 +14,7 @@ require ( github.com/kdomanski/iso9660 v0.4.0 github.com/klauspost/cpuid/v2 v2.3.0 github.com/mulgadc/predastore v1.7.1-0.20260612051752-c013cb87f908 - github.com/mulgadc/viperblock v1.7.1-0.20260610024504-5344356f78a2 + github.com/mulgadc/viperblock v1.7.1-0.20260615121352-3ca32581b809 github.com/nats-io/nats-server/v2 v2.14.2 github.com/nats-io/nats.go v1.52.0 github.com/ovn-kubernetes/libovsdb v0.8.1 diff --git a/go.sum b/go.sum index dfac33340..03e461e59 100644 --- a/go.sum +++ b/go.sum @@ -290,8 +290,8 @@ github.com/muesli/termenv v0.16.0 h1:S5AlUN9dENB57rsbnkPyfdGuWIlkmzJjbFf0Tf5FWUc github.com/muesli/termenv v0.16.0/go.mod h1:ZRfOIKPFDYQoDFF4Olj7/QJbW60Ol/kL1pU3VfY/Cnk= github.com/mulgadc/predastore v1.7.1-0.20260612051752-c013cb87f908 h1:YJR9m17PO1ElOy+GfBg+izTy+h4v+6gZQh9/6EIUWZQ= github.com/mulgadc/predastore v1.7.1-0.20260612051752-c013cb87f908/go.mod h1:0334T4L4Dby54TCuugpumyOhtvHQX+WSxQ/REDNSk9o= -github.com/mulgadc/viperblock v1.7.1-0.20260610024504-5344356f78a2 h1:r3GKRUntS3hXwTNBAYIhHsHUP0A/12FzH3g58bHd0Zc= -github.com/mulgadc/viperblock v1.7.1-0.20260610024504-5344356f78a2/go.mod h1:F/LOVydipnyjX9QMkAuf/gjdWeYjaniGHU4emwJwq+U= +github.com/mulgadc/viperblock v1.7.1-0.20260615121352-3ca32581b809 h1:jYcLBJGonjlMeHlR5P70AUC3Miw1U0cYs73vZ0qiFDw= +github.com/mulgadc/viperblock v1.7.1-0.20260615121352-3ca32581b809/go.mod h1:F/LOVydipnyjX9QMkAuf/gjdWeYjaniGHU4emwJwq+U= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= diff --git a/spinifex/handlers/ec2/snapshot/service_impl.go b/spinifex/handlers/ec2/snapshot/service_impl.go index dad9bddad..c6c7a4a31 100644 --- a/spinifex/handlers/ec2/snapshot/service_impl.go +++ b/spinifex/handlers/ec2/snapshot/service_impl.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "io" "log/slog" "maps" "strconv" @@ -209,8 +210,17 @@ func (s *SnapshotServiceImpl) CreateSnapshot(input *ec2.CreateSnapshotInput, acc } defer volumeResult.Body.Close() + volumeBody, err := io.ReadAll(volumeResult.Body) + if err != nil { + slog.Error("CreateSnapshot failed to read volume config", "volumeId", volumeID, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + + // config.json may be an at-rest encryption envelope; StateBody unwraps it to + // the inner VBState. Decoding the raw envelope yields a zero state + // (SizeGiB==0), which the size guard below would reject as a 500. var volumeState viperblock.VBState - if err := json.NewDecoder(volumeResult.Body).Decode(&volumeState); err != nil { + if err := json.Unmarshal(viperblock.StateBody(volumeBody), &volumeState); err != nil { slog.Error("CreateSnapshot failed to decode volume config", "volumeId", volumeID, "err", err) return nil, errors.New(awserrors.ErrorServerInternal) } @@ -447,10 +457,15 @@ func (s *SnapshotServiceImpl) snapshotInUseByVolumes(snapshotID string) (bool, e continue // volume may not have a config yet } - var state viperblock.VBState - decodeErr := json.NewDecoder(result.Body).Decode(&state) + scanBody, readErr := io.ReadAll(result.Body) _ = result.Body.Close() - if decodeErr != nil { + if readErr != nil { + continue + } + // Unwrap the encryption envelope so encrypted volumes are scanned too; + // a raw decode yields a zero state and silently drops their snapshots. + var state viperblock.VBState + if decodeErr := json.Unmarshal(viperblock.StateBody(scanBody), &state); decodeErr != nil { continue } diff --git a/spinifex/handlers/ec2/snapshot/service_impl_test.go b/spinifex/handlers/ec2/snapshot/service_impl_test.go index 501950265..132b3284f 100644 --- a/spinifex/handlers/ec2/snapshot/service_impl_test.go +++ b/spinifex/handlers/ec2/snapshot/service_impl_test.go @@ -135,6 +135,85 @@ func TestCreateSnapshot_VolumeZeroSize(t *testing.T) { assert.Contains(t, err.Error(), awserrors.ErrorServerInternal) } +// createEncryptedTestVolume seeds config.json as an at-rest encryption envelope +// ({payload, authtag}) wrapping a VBState, matching what an encrypted volume +// persists. The snapshot handler must unwrap it via StateBody, not decode raw. +func createEncryptedTestVolume(t *testing.T, store *objectstore.MemoryObjectStore, volumeID string, sizeGiB int) { + inner := viperblock.VBState{ + EncryptionEnabled: true, + VolumeConfig: viperblock.VolumeConfig{ + VolumeMetadata: viperblock.VolumeMetadata{ + SizeGiB: uint64(sizeGiB), + AvailabilityZone: "us-east-1a", + }, + }, + } + payload, err := json.Marshal(inner) + require.NoError(t, err) + + envelope, err := json.Marshal(map[string]any{ + "payload": json.RawMessage(payload), + "authtag": "deadbeefdeadbeefdeadbeefdeadbeef", + }) + require.NoError(t, err) + + _, err = store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String(volumeID + "/config.json"), + Body: strings.NewReader(string(envelope)), + }) + require.NoError(t, err) +} + +// TestCreateSnapshot_EncryptedVolumeEnvelope verifies an encrypted volume (whose +// config.json is an envelope) snapshots successfully instead of 500ing on a +// zero-decoded VBState. +func TestCreateSnapshot_EncryptedVolumeEnvelope(t *testing.T) { + svc, store := setupTestSnapshotService(t) + + volumeID := "vol-enc-snap" + createEncryptedTestVolume(t, store, volumeID, 100) + + result, err := svc.CreateSnapshot(&ec2.CreateSnapshotInput{ + VolumeId: aws.String(volumeID), + }, testAccountID) + require.NoError(t, err) + require.NotNil(t, result) + assert.Equal(t, int64(100), *result.VolumeSize) + assert.True(t, *result.Encrypted) +} + +// TestSnapshotInUseByVolumes_EncryptedVolume verifies the membership scan sees an +// encrypted volume's source SnapshotID through the envelope (a raw decode would +// miss it, letting an in-use snapshot be deleted). +func TestSnapshotInUseByVolumes_EncryptedVolume(t *testing.T) { + svc, store := setupTestSnapshotService(t) + + inner := viperblock.VBState{ + EncryptionEnabled: true, + VolumeConfig: viperblock.VolumeConfig{ + VolumeMetadata: viperblock.VolumeMetadata{SizeGiB: 50, SnapshotID: "snap-src"}, + }, + } + payload, err := json.Marshal(inner) + require.NoError(t, err) + envelope, err := json.Marshal(map[string]any{ + "payload": json.RawMessage(payload), + "authtag": "deadbeefdeadbeefdeadbeefdeadbeef", + }) + require.NoError(t, err) + _, err = store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String("vol-fromsnap/config.json"), + Body: strings.NewReader(string(envelope)), + }) + require.NoError(t, err) + + inUse, err := svc.snapshotInUseByVolumes("snap-src") + require.NoError(t, err) + assert.True(t, inUse, "encrypted volume's SnapshotID must be visible through the envelope") +} + // TestCreateSnapshot_VolumeNotFound tests creating a snapshot from non-existent volume func TestCreateSnapshot_VolumeNotFound(t *testing.T) { svc, _ := setupTestSnapshotService(t) diff --git a/spinifex/handlers/ec2/volume/config_route_test.go b/spinifex/handlers/ec2/volume/config_route_test.go new file mode 100644 index 000000000..990038d1e --- /dev/null +++ b/spinifex/handlers/ec2/volume/config_route_test.go @@ -0,0 +1,168 @@ +package handlers_ec2_volume + +import ( + "encoding/json" + "io" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/s3" + "github.com/mulgadc/spinifex/spinifex/objectstore" + "github.com/mulgadc/spinifex/spinifex/types" + "github.com/mulgadc/viperblock/viperblock" + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// startTestNATS spins up an in-process NATS server and returns a connected client. +func startTestNATS(t *testing.T) *nats.Conn { + 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() }) + return nc +} + +// seedEncryptedConfig writes a sealed (EncryptionEnabled=true) VBState to the store. +func seedEncryptedConfig(t *testing.T, store *objectstore.MemoryObjectStore, volumeID string) { + t.Helper() + state := viperblock.VBState{ + VolumeName: volumeID, + VolumeSize: 10 * 1024 * 1024 * 1024, + BlockSize: 4096, + EncryptionEnabled: true, + VolumeConfig: viperblock.VolumeConfig{ + VolumeMetadata: viperblock.VolumeMetadata{VolumeID: volumeID, SizeGiB: 10, State: "available"}, + }, + } + data, err := json.Marshal(state) + require.NoError(t, err) + _, err = store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String(volumeID + "/config.json"), + Body: strings.NewReader(string(data)), + }) + require.NoError(t, err) +} + +// getStoredConfig reads the raw config.json bytes from the memory store. +func getStoredConfig(t *testing.T, store *objectstore.MemoryObjectStore, volumeID string) []byte { + t.Helper() + res, err := store.GetObject(&s3.GetObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String(volumeID + "/config.json"), + }) + require.NoError(t, err) + defer res.Body.Close() + data, err := io.ReadAll(res.Body) + require.NoError(t, err) + return data +} + +// TestUpdateVolumeState_EncryptedRoutesToLiveVB verifies an encrypted-volume +// state update is delegated to the owning node via ebs.config.{volumeID} rather +// than rewriting the sealed object directly. +func TestUpdateVolumeState_EncryptedRoutesToLiveVB(t *testing.T) { + store := objectstore.NewMemoryObjectStore() + svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) + svc.natsConn = startTestNATS(t) + + volumeID := "vol-enc-live" + seedEncryptedConfig(t, store, volumeID) + + var got atomic.Pointer[types.EBSConfigUpdateRequest] + sub, err := svc.natsConn.Subscribe("ebs.config."+volumeID, func(msg *nats.Msg) { + var req types.EBSConfigUpdateRequest + _ = json.Unmarshal(msg.Data, &req) + got.Store(&req) + data, _ := json.Marshal(types.EBSConfigUpdateResponse{Volume: volumeID, Success: true}) + _ = msg.Respond(data) + }) + require.NoError(t, err) + defer sub.Unsubscribe() + + err = svc.UpdateVolumeState(volumeID, "in-use", "i-abc", "/dev/nbd0") + require.NoError(t, err) + + req := got.Load() + require.NotNil(t, req, "per-volume responder must receive the config update") + var vc viperblock.VolumeConfig + require.NoError(t, json.Unmarshal(req.VolumeConfig, &vc)) + assert.Equal(t, "in-use", vc.VolumeMetadata.State) + assert.Equal(t, "i-abc", vc.VolumeMetadata.AttachedInstance) + assert.Equal(t, "/dev/nbd0", vc.VolumeMetadata.DeviceName) + + // The sealed object must NOT have been rewritten by the handler. + raw := getStoredConfig(t, store, volumeID) + var still viperblock.VBState + require.NoError(t, json.Unmarshal(viperblock.StateBody(raw), &still)) + assert.True(t, still.EncryptionEnabled) + assert.Equal(t, "available", still.VolumeConfig.VolumeMetadata.State, + "handler must not mutate the sealed object directly; the keyholder owns it") +} + +// TestUpdateVolumeState_EncryptedDetachedFallsBackToQueueGroup verifies that with +// no per-volume responder (volume unmounted), the update routes to the ebs.config +// queue group. +func TestUpdateVolumeState_EncryptedDetachedFallsBackToQueueGroup(t *testing.T) { + store := objectstore.NewMemoryObjectStore() + svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) + svc.natsConn = startTestNATS(t) + + volumeID := "vol-enc-detached" + seedEncryptedConfig(t, store, volumeID) + + var hits atomic.Int32 + sub, err := svc.natsConn.QueueSubscribe("ebs.config", "spinifex-workers", func(msg *nats.Msg) { + hits.Add(1) + data, _ := json.Marshal(types.EBSConfigUpdateResponse{Volume: volumeID, Success: true}) + _ = msg.Respond(data) + }) + require.NoError(t, err) + defer sub.Unsubscribe() + + err = svc.UpdateVolumeState(volumeID, "available", "", "") + require.NoError(t, err) + assert.Equal(t, int32(1), hits.Load(), "detached encrypted update must hit the ebs.config queue group") +} + +// TestUpdateVolumeState_EncryptedKeyholderError surfaces a keyholder failure +// instead of silently succeeding. +func TestUpdateVolumeState_EncryptedKeyholderError(t *testing.T) { + store := objectstore.NewMemoryObjectStore() + svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) + svc.natsConn = startTestNATS(t) + + volumeID := "vol-enc-err" + seedEncryptedConfig(t, store, volumeID) + + sub, err := svc.natsConn.Subscribe("ebs.config."+volumeID, func(msg *nats.Msg) { + data, _ := json.Marshal(types.EBSConfigUpdateResponse{Volume: volumeID, Error: "reseal failed"}) + _ = msg.Respond(data) + }) + require.NoError(t, err) + defer sub.Unsubscribe() + + err = svc.UpdateVolumeState(volumeID, "in-use", "i-x", "/dev/nbd0") + require.Error(t, err) + assert.Contains(t, err.Error(), "reseal failed") +} diff --git a/spinifex/handlers/ec2/volume/service_impl.go b/spinifex/handlers/ec2/volume/service_impl.go index a11a39403..8c5f91e58 100644 --- a/spinifex/handlers/ec2/volume/service_impl.go +++ b/spinifex/handlers/ec2/volume/service_impl.go @@ -1026,6 +1026,18 @@ func (s *VolumeServiceImpl) getVolumeConfigAndEncryption(volumeID string) (*vipe func (s *VolumeServiceImpl) putVolumeConfig(volumeID string, cfg *viperblock.VolumeConfig) error { configKey := volumeID + "/config.json" + // config.json for an encrypted volume is a sealed VBState whose AES-GCM tag + // and StateSeqNum-derived nonce can only be advanced by the master-key holder + // that owns the volume. Rewriting it here would strip the tag or — racing the + // live VB — reuse a nonce, so route the update through viperblockd instead. + encrypted, err := s.configIsEncrypted(volumeID) + if err != nil { + return err + } + if encrypted { + return s.putVolumeConfigViaKeyholder(volumeID, cfg) + } + data, err := s.mergeVolumeConfig(configKey, cfg) if err != nil { return err @@ -1043,6 +1055,72 @@ func (s *VolumeServiceImpl) putVolumeConfig(volumeID string, cfg *viperblock.Vol return nil } +// configIsEncrypted reports whether the persisted config.json for volumeID is a +// sealed (encrypted) VBState. A missing object (new volume) reports false. +func (s *VolumeServiceImpl) configIsEncrypted(volumeID string) (bool, error) { + getResult, err := s.store.GetObject(&s3.GetObjectInput{ + Bucket: aws.String(s.bucketName), + Key: aws.String(volumeID + "/config.json"), + }) + if err != nil { + if objectstore.IsNoSuchKeyError(err) { + return false, nil + } + return false, fmt.Errorf("failed to read config for encryption check: %w", err) + } + defer getResult.Body.Close() + + body, err := io.ReadAll(getResult.Body) + if err != nil { + return false, fmt.Errorf("failed to read config body: %w", err) + } + + var state viperblock.VBState + if decodeErr := json.Unmarshal(viperblock.StateBody(body), &state); decodeErr == nil && state.BlockSize != 0 { + return state.EncryptionEnabled, nil + } + return false, nil +} + +// putVolumeConfigViaKeyholder routes an encrypted-volume config update through +// viperblockd, the master-key holder. The owning node answers +// ebs.config.{volumeID} and updates its live VB. If no node owns the volume +// (detached), ErrNoResponders routes to the ebs.config queue group, where a +// worker opens the volume exclusively and reseals. A detached volume has no +// concurrent writer, so the reopen is nonce-safe. +func (s *VolumeServiceImpl) putVolumeConfigViaKeyholder(volumeID string, cfg *viperblock.VolumeConfig) error { + if s.natsConn == nil { + return fmt.Errorf("encrypted volume %s requires NATS to reach the viperblock keyholder, but no connection is configured", volumeID) + } + + cfgData, err := json.Marshal(cfg) + if err != nil { + return fmt.Errorf("marshal VolumeConfig: %w", err) + } + reqData, err := json.Marshal(types.EBSConfigUpdateRequest{Volume: volumeID, VolumeConfig: cfgData}) + if err != nil { + return fmt.Errorf("marshal config update request: %w", err) + } + + msg, err := s.natsConn.Request("ebs.config."+volumeID, reqData, 30*time.Second) + if errors.Is(err, nats.ErrNoResponders) { + // Volume not mounted anywhere -- fall back to the queue group. + msg, err = s.natsConn.Request("ebs.config", reqData, 30*time.Second) + } + if err != nil { + return fmt.Errorf("ebs.config request for %s: %w", volumeID, err) + } + + var resp types.EBSConfigUpdateResponse + if err := json.Unmarshal(msg.Data, &resp); err != nil { + return fmt.Errorf("unmarshal config update response: %w", err) + } + if !resp.Success || resp.Error != "" { + return fmt.Errorf("ebs.config update for %s failed: %s", volumeID, resp.Error) + } + return nil +} + // mergeVolumeConfig reads existing config.json from S3 and merges the new VolumeConfig // into it, preserving full VBState when present. Refuses to merge encrypted VBState — // re-marshaling without the master key would corrupt the on-disk AES-GCM tag. diff --git a/spinifex/nbd/nbd.go b/spinifex/nbd/nbd.go index 707bc0865..236a2c94f 100644 --- a/spinifex/nbd/nbd.go +++ b/spinifex/nbd/nbd.go @@ -25,6 +25,11 @@ type NBDKitConfig struct { CacheSize int `json:"cache_size"` ShardWAL bool `json:"shardwal"` // Enable sharded WAL (default false) UseTCP bool `json:"use_tcp"` // If true, use TCP transport; otherwise use Unix socket + + // EncryptionKeyFile is the path to the shared AES-256 master key. When set, + // it is forwarded to the viperblock plugin so encrypted volumes open with + // matching encryption state. Empty → plugin opens in cleartext mode. + EncryptionKeyFile string `json:"encryption_key_file"` } // buildArgs constructs the nbdkit command-line arguments from the config. @@ -66,6 +71,12 @@ func (cfg *NBDKitConfig) buildArgs() ([]string, error) { fmt.Sprintf("shardwal=%t", cfg.ShardWAL), } + // Only forward the key when configured; an empty value would explicitly + // set the plugin to cleartext and override its ENCRYPTION_KEY_FILE fallback. + if cfg.EncryptionKeyFile != "" { + pluginArgs = append(pluginArgs, fmt.Sprintf("encryption_key_file=%s", cfg.EncryptionKeyFile)) + } + args = append(args, pluginArgs...) return args, nil } diff --git a/spinifex/nbd/nbd_test.go b/spinifex/nbd/nbd_test.go index a5111b04e..521c89aec 100644 --- a/spinifex/nbd/nbd_test.go +++ b/spinifex/nbd/nbd_test.go @@ -233,6 +233,44 @@ func TestBuildArgs_ArgOrdering(t *testing.T) { } } +func TestBuildArgs_EncryptionKeyFileForwarded(t *testing.T) { + cfg := &NBDKitConfig{ + Socket: "/tmp/nbd.sock", + PidFile: "/tmp/nbd.pid", + PluginPath: "/plugin.so", + EncryptionKeyFile: "/etc/spinifex/viperblock/encryption.key", + } + + args, err := cfg.buildArgs() + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + want := "encryption_key_file=/etc/spinifex/viperblock/encryption.key" + if indexOf(args, want) < 0 { + t.Errorf("expected %q in args, got: %v", want, args) + } +} + +func TestBuildArgs_EncryptionKeyFileOmittedWhenEmpty(t *testing.T) { + cfg := &NBDKitConfig{ + Socket: "/tmp/nbd.sock", + PidFile: "/tmp/nbd.pid", + PluginPath: "/plugin.so", + } + + args, err := cfg.buildArgs() + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + for _, arg := range args { + if len(arg) >= 19 && arg[:19] == "encryption_key_file" { + t.Errorf("encryption_key_file must be absent when unset, got: %q", arg) + } + } +} + func assertArgs(t *testing.T, expected, got []string) { t.Helper() if len(expected) != len(got) { diff --git a/spinifex/services/viperblockd/config_update_test.go b/spinifex/services/viperblockd/config_update_test.go new file mode 100644 index 000000000..1ff5fd95e --- /dev/null +++ b/spinifex/services/viperblockd/config_update_test.go @@ -0,0 +1,246 @@ +package viperblockd + +// Tests for the encrypted-volume config-update path: applyConfigUpdate, +// makeConfigUpdateHandler (ebs.config.{volumeID}), and the ebs.config +// queue-group fallback handler in launchService. + +import ( + "encoding/json" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/types" + "github.com/mulgadc/viperblock/viperblock" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const gib = uint64(1024 * 1024 * 1024) + +// configReq marshals a VolumeConfig carrying the given size into an update request. +func configReq(t *testing.T, volume string, sizeGiB uint64) types.EBSConfigUpdateRequest { + t.Helper() + vc := viperblock.VolumeConfig{} + vc.VolumeMetadata.VolumeID = volume + vc.VolumeMetadata.SizeGiB = sizeGiB + raw, err := json.Marshal(vc) + require.NoError(t, err) + return types.EBSConfigUpdateRequest{Volume: volume, VolumeConfig: raw} +} + +// --- applyConfigUpdate --- + +func TestApplyConfigUpdate_GrowsVolumeSize(t *testing.T) { + t.Parallel() + vb := createTestVBWithState(t, "vol-apply-grow") + + require.NoError(t, applyConfigUpdate(vb, configReq(t, "vol-apply-grow", 5))) + + assert.Equal(t, uint64(5), vb.VolumeConfig.VolumeMetadata.SizeGiB) + assert.Equal(t, 5*gib, vb.VolumeSize) +} + +func TestApplyConfigUpdate_ShrinkKeepsSize(t *testing.T) { + t.Parallel() + vb := createTestVBWithState(t, "vol-apply-shrink") + vb.VolumeSize = 10 * gib + + require.NoError(t, applyConfigUpdate(vb, configReq(t, "vol-apply-shrink", 2))) + + // Config metadata applied, but VolumeSize is grow-only. + assert.Equal(t, uint64(2), vb.VolumeConfig.VolumeMetadata.SizeGiB) + assert.Equal(t, 10*gib, vb.VolumeSize) +} + +func TestApplyConfigUpdate_InvalidVolumeConfig(t *testing.T) { + t.Parallel() + vb := createTestVBWithState(t, "vol-apply-bad") + + req := types.EBSConfigUpdateRequest{Volume: "vol-apply-bad", VolumeConfig: json.RawMessage(`"not-an-object"`)} + err := applyConfigUpdate(vb, req) + require.Error(t, err) + assert.Contains(t, err.Error(), "unmarshal VolumeConfig") +} + +// --- makeConfigUpdateHandler (ebs.config.{volumeID}) --- + +func TestConfigUpdateHandler_Success(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + vb := createTestVBWithState(t, "vol-cfg-ok") + sub, err := nc.Subscribe("ebs.config.vol-cfg-ok", makeConfigUpdateHandler(vb, "vol-cfg-ok")) + require.NoError(t, err) + defer sub.Unsubscribe() + nc.Flush() + + reqData, _ := json.Marshal(configReq(t, "vol-cfg-ok", 7)) + msg, err := nc.Request("ebs.config.vol-cfg-ok", reqData, 3*time.Second) + require.NoError(t, err) + + var resp types.EBSConfigUpdateResponse + require.NoError(t, json.Unmarshal(msg.Data, &resp)) + assert.True(t, resp.Success) + assert.Empty(t, resp.Error) + assert.Equal(t, uint64(7), vb.VolumeConfig.VolumeMetadata.SizeGiB) +} + +func TestConfigUpdateHandler_InvalidJSON(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + vb := createTestVBWithState(t, "vol-cfg-badjson") + sub, err := nc.Subscribe("ebs.config.vol-cfg-badjson", makeConfigUpdateHandler(vb, "vol-cfg-badjson")) + require.NoError(t, err) + defer sub.Unsubscribe() + nc.Flush() + + msg, err := nc.Request("ebs.config.vol-cfg-badjson", []byte("not json {{{"), 3*time.Second) + require.NoError(t, err) + + var resp types.EBSConfigUpdateResponse + require.NoError(t, json.Unmarshal(msg.Data, &resp)) + assert.False(t, resp.Success) + assert.Contains(t, resp.Error, "bad request:") +} + +func TestConfigUpdateHandler_ApplyError(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + vb := createTestVBWithState(t, "vol-cfg-applyerr") + sub, err := nc.Subscribe("ebs.config.vol-cfg-applyerr", makeConfigUpdateHandler(vb, "vol-cfg-applyerr")) + require.NoError(t, err) + defer sub.Unsubscribe() + nc.Flush() + + reqData, _ := json.Marshal(types.EBSConfigUpdateRequest{ + Volume: "vol-cfg-applyerr", + VolumeConfig: json.RawMessage(`"not-an-object"`), + }) + msg, err := nc.Request("ebs.config.vol-cfg-applyerr", reqData, 3*time.Second) + require.NoError(t, err) + + var resp types.EBSConfigUpdateResponse + require.NoError(t, json.Unmarshal(msg.Data, &resp)) + assert.False(t, resp.Success) + assert.Contains(t, resp.Error, "unmarshal VolumeConfig") +} + +// --- ebs.config queue-group fallback --- + +func TestEBSConfigQueueGroup_LiveVB(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + vb := createTestVBWithState(t, "vol-qg-live") + cfg := setupTestConfig(t, natsURL) + cfg.MountedVolumes = []MountedVolume{{Name: "vol-qg-live", VB: vb}} + + go func() { launchService(cfg) }() + time.Sleep(500 * time.Millisecond) + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + reqData, _ := json.Marshal(configReq(t, "vol-qg-live", 9)) + msg, err := nc.Request("ebs.config", reqData, 3*time.Second) + require.NoError(t, err) + + var resp types.EBSConfigUpdateResponse + require.NoError(t, json.Unmarshal(msg.Data, &resp)) + assert.True(t, resp.Success) + assert.Empty(t, resp.Error) + assert.Equal(t, uint64(9), vb.VolumeConfig.VolumeMetadata.SizeGiB) +} + +func TestEBSConfigQueueGroup_InvalidJSON(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + cfg := setupTestConfig(t, natsURL) + + go func() { launchService(cfg) }() + time.Sleep(500 * time.Millisecond) + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + msg, err := nc.Request("ebs.config", []byte("not json {{{"), 3*time.Second) + require.NoError(t, err) + + var resp types.EBSConfigUpdateResponse + require.NoError(t, json.Unmarshal(msg.Data, &resp)) + assert.False(t, resp.Success) + assert.Contains(t, resp.Error, "bad request:") +} + +func TestEBSConfigQueueGroup_DetachedOpenFails(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + // No mounted volume, so the queue-group takes the detached branch and tries + // openVolumeVB against the mock S3 host, which fails. + cfg := setupTestConfig(t, natsURL) + + go func() { launchService(cfg) }() + time.Sleep(500 * time.Millisecond) + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + reqData, _ := json.Marshal(configReq(t, "vol-qg-detached", 3)) + msg, err := nc.Request("ebs.config", reqData, 5*time.Second) + require.NoError(t, err) + + var resp types.EBSConfigUpdateResponse + require.NoError(t, json.Unmarshal(msg.Data, &resp)) + assert.False(t, resp.Success) + assert.Contains(t, resp.Error, "open volume:") +} diff --git a/spinifex/services/viperblockd/viperblockd.go b/spinifex/services/viperblockd/viperblockd.go index 7ccf33233..be03c28e2 100644 --- a/spinifex/services/viperblockd/viperblockd.go +++ b/spinifex/services/viperblockd/viperblockd.go @@ -75,6 +75,7 @@ type MountedVolume struct { PID int VB *viperblock.VB // Reference to viperblock instance for state sync/flush SnapshotSub *nats.Subscription // Per-volume snapshot subscription (ebs.snapshot.{volumeID}) + ConfigSub *nats.Subscription // Per-volume config-update subscription (ebs.config.{volumeID}) } type Config struct { @@ -157,6 +158,79 @@ func makeSnapshotHandler(vb *viperblock.VB, volumeName string) nats.MsgHandler { } } +// applyConfigUpdate writes a control-plane VolumeConfig onto a viperblock +// instance and reseals its state. For encrypted volumes SaveState recomputes the +// AES-GCM tag under the volume's current StateSeqNum, so the caller MUST own the +// volume exclusively (live mounted VB, or a freshly opened detached one) to keep +// the GCM nonce unique. +func applyConfigUpdate(vb *viperblock.VB, req types.EBSConfigUpdateRequest) error { + var vc viperblock.VolumeConfig + if err := json.Unmarshal(req.VolumeConfig, &vc); err != nil { + return fmt.Errorf("unmarshal VolumeConfig: %w", err) + } + vb.VolumeConfig = vc + // Reconcile grow-only volume size (mirrors the EC2 handler merge path). + if sz := vc.VolumeMetadata.SizeGiB * 1024 * 1024 * 1024; sz > vb.VolumeSize { + vb.VolumeSize = sz + } + return vb.SaveState() +} + +// makeConfigUpdateHandler returns a NATS handler for volume-specific config +// updates (ebs.config.{volumeID}). It runs against the live mounted VB, which is +// the single writer that owns the volume's StateSeqNum. +func makeConfigUpdateHandler(vb *viperblock.VB, volumeName string) nats.MsgHandler { + return func(msg *nats.Msg) { + var req types.EBSConfigUpdateRequest + if err := json.Unmarshal(msg.Data, &req); err != nil { + slog.Error("Failed to unmarshal ebs.config message", "volume", volumeName, "err", err) + respondJSON(msg, types.EBSConfigUpdateResponse{Volume: volumeName, Error: fmt.Sprintf("bad request: %v", err)}) + return + } + if err := applyConfigUpdate(vb, req); err != nil { + slog.Error("ebs.config: live VB update failed", "volume", volumeName, "err", err) + respondJSON(msg, types.EBSConfigUpdateResponse{Volume: volumeName, Error: err.Error()}) + return + } + slog.Info("ebs.config: live VB state updated", "volume", volumeName) + respondJSON(msg, types.EBSConfigUpdateResponse{Volume: volumeName, Success: true}) + } +} + +// openVolumeVB opens an existing viperblock volume for a short-lived, exclusive +// state update (detached-volume config reseal). The caller MUST Close the +// returned VB. Construction mirrors the ebs.mount path so encrypted volumes open +// with the master key and matching encryption state. +func openVolumeVB(cfg *Config, volumeName string) (*viperblock.VB, error) { + s3cfg := s3.S3Config{ + VolumeName: volumeName, + Bucket: cfg.Bucket, + Region: cfg.Region, + AccessKey: cfg.AccessKey, + SecretKey: cfg.SecretKey, + Host: admin.DialTarget(cfg.S3Host), + } + vbconfig := viperblock.VB{ + VolumeName: volumeName, + VolumeSize: 1, // Recalculated on LoadState. + BaseDir: cfg.BaseDir, + VolumeConfig: viperblock.VolumeConfig{}, + MasterKey: cfg.masterKey, + EncryptionEnabled: cfg.masterKey != nil, + } + vb, err := viperblock.New(&vbconfig, "s3", s3cfg) + if err != nil { + return nil, fmt.Errorf("new viperblock: %w", err) + } + if err := vb.Backend.Init(); err != nil { + return nil, fmt.Errorf("backend init: %w", err) + } + if err := loadStateWithRetry(vb, volumeName); err != nil { + return nil, fmt.Errorf("load state: %w", err) + } + return vb, nil +} + // respondJSON marshals data and sends it as a NATS response. On marshal // failure a raw JSON error string is sent instead. func respondJSON(msg *nats.Msg, data any) { @@ -277,6 +351,12 @@ func launchService(cfg *Config) (err error) { slog.Error("Failed to unsubscribe snapshot topic", "volume", ebsRequest.Volume, "err", err) } } + // Unsubscribe from volume-specific config-update topic + if matched.ConfigSub != nil { + if err := matched.ConfigSub.Unsubscribe(); err != nil { + slog.Error("Failed to unsubscribe config topic", "volume", ebsRequest.Volume, "err", err) + } + } // Stop WAL syncer and kill nbdkit process if matched.VB != nil { matched.VB.StopWALSyncer() @@ -355,6 +435,13 @@ func launchService(cfg *Config) (err error) { } } + // Unsubscribe from volume-specific config-update topic + if matched.ConfigSub != nil { + if err := matched.ConfigSub.Unsubscribe(); err != nil { + slog.Error("Failed to unsubscribe config topic", "volume", ebsRequest.Name, "err", err) + } + } + // Clean up the VB instance's background goroutine. // This VB is state-only (LoadState/sync) — actual I/O is in the nbdkit plugin process. if matched.VB != nil { @@ -426,6 +513,61 @@ func launchService(cfg *Config) (err error) { return fmt.Errorf("failed to subscribe to ebs.sync: %w", err) } + // ebs.config is the fallback for encrypted-volume config updates whose + // per-volume ebs.config.{volumeID} topic had no responder (volume not + // mounted anywhere). A detached volume has no live writer, so any worker may + // open it exclusively and reseal. A mount that raced in is still handled by + // preferring the live VB when this node happens to own it. + if _, err := nc.QueueSubscribe("ebs.config", "spinifex-workers", func(msg *nats.Msg) { + var req types.EBSConfigUpdateRequest + if err := json.Unmarshal(msg.Data, &req); err != nil { + slog.Error("Failed to unmarshal ebs.config message", "err", err) + respondJSON(msg, types.EBSConfigUpdateResponse{Error: fmt.Sprintf("bad request: %v", err)}) + return + } + + cfg.mu.Lock() + var live *viperblock.VB + for _, volume := range cfg.MountedVolumes { + if volume.Name == req.Volume && volume.VB != nil { + live = volume.VB + break + } + } + cfg.mu.Unlock() + + if live != nil { + if err := applyConfigUpdate(live, req); err != nil { + slog.Error("ebs.config: live VB update failed", "volume", req.Volume, "err", err) + respondJSON(msg, types.EBSConfigUpdateResponse{Volume: req.Volume, Error: err.Error()}) + return + } + slog.Info("ebs.config: live VB state updated (fallback path)", "volume", req.Volume) + respondJSON(msg, types.EBSConfigUpdateResponse{Volume: req.Volume, Success: true}) + return + } + + vb, err := openVolumeVB(cfg, req.Volume) + if err != nil { + slog.Error("ebs.config: failed to open detached volume", "volume", req.Volume, "err", err) + respondJSON(msg, types.EBSConfigUpdateResponse{Volume: req.Volume, Error: fmt.Sprintf("open volume: %v", err)}) + return + } + applyErr := applyConfigUpdate(vb, req) + if closeErr := vb.Close(); closeErr != nil { + slog.Error("ebs.config: VB close failed", "volume", req.Volume, "err", closeErr) + } + if applyErr != nil { + slog.Error("ebs.config: detached volume update failed", "volume", req.Volume, "err", applyErr) + respondJSON(msg, types.EBSConfigUpdateResponse{Volume: req.Volume, Error: applyErr.Error()}) + return + } + slog.Info("ebs.config: detached volume state updated", "volume", req.Volume) + respondJSON(msg, types.EBSConfigUpdateResponse{Volume: req.Volume, Success: true}) + }); err != nil { + return fmt.Errorf("failed to subscribe to ebs.config: %w", err) + } + // Note: ebs.snapshot is handled per-volume via ebs.snapshot.{volumeID} topics, // subscribed at mount time and unsubscribed at unmount time. This ensures // snapshot requests are routed to the node that owns the volume. @@ -596,6 +738,8 @@ func launchService(cfg *Config) (err error) { SecretKey: cfg.SecretKey, CacheSize: nbdCacheSize, ShardWAL: cfg.ShardWAL, + + EncryptionKeyFile: cfg.EncryptionKeyFile, } // Create a unique error channel for this specific mount request @@ -672,6 +816,13 @@ func launchService(cfg *Config) (err error) { slog.Error("Failed to subscribe to volume snapshot topic", "volume", ebsRequest.Name, "err", err) } + // Subscribe to volume-specific config-update topic so encrypted-volume + // metadata writes route to this node's live VB (the StateSeqNum owner). + configSub, err := nc.Subscribe(fmt.Sprintf("ebs.config.%s", ebsRequest.Name), makeConfigUpdateHandler(vb, ebsRequest.Name)) + if err != nil { + slog.Error("Failed to subscribe to volume config topic", "volume", ebsRequest.Name, "err", err) + } + cfg.mu.Lock() cfg.MountedVolumes = append(cfg.MountedVolumes, MountedVolume{ Name: ebsRequest.Name, @@ -681,6 +832,7 @@ func launchService(cfg *Config) (err error) { PID: pid, VB: vb, SnapshotSub: snapSub, + ConfigSub: configSub, }) cfg.mu.Unlock() diff --git a/spinifex/types/ebs.go b/spinifex/types/ebs.go index 11ab65786..8ad217f01 100644 --- a/spinifex/types/ebs.go +++ b/spinifex/types/ebs.go @@ -1,6 +1,9 @@ package types -import "sync" +import ( + "encoding/json" + "sync" +) type EBSRequests struct { Requests []EBSRequest `json:"Requests" mapstructure:"ebs_requests"` @@ -70,3 +73,19 @@ type EBSSnapshotResponse struct { Success bool `json:"Success"` Error string `json:"Error"` } + +// EBSConfigUpdateRequest carries a control-plane VolumeConfig update for an +// encrypted volume. config.json is a sealed VBState; only the master-key holder +// (viperblockd) can reseal it, so the EC2 edge ships the new config here instead +// of rewriting the object directly. VolumeConfig is a marshaled +// viperblock.VolumeConfig (RawMessage keeps this package dependency-free). +type EBSConfigUpdateRequest struct { + Volume string `json:"Volume"` + VolumeConfig json.RawMessage `json:"VolumeConfig"` +} + +type EBSConfigUpdateResponse struct { + Volume string `json:"Volume"` + Success bool `json:"Success"` + Error string `json:"Error"` +}