snapshot: harden object offload validation

cmd/elastickv-snapshot-offload/main.go

@@ -26,6 +26,7 @@ const (
 	commandRestore = "restore"
 	storeLocal     = "local"
 	storeS3        = "s3"
+	s3SSEAES256    = "AES256"
 	s3SSEAWSKMS    = "aws:kms"
 )
 
@@ -177,15 +178,16 @@ func addStoreFlags(fs *flag.FlagSet, cfg *storeFlags) {
 	cfg.storeKind = storeLocal
 	cfg.s3Region = "us-east-1"
 	cfg.s3PathStyle = true
+	cfg.s3ServerSideEncryption = string(s3SSEAES256)
 	fs.StringVar(&cfg.storeKind, "store", storeLocal, "Object store backend: local or s3")
 	fs.StringVar(&cfg.localRoot, "local-root", "", "Local object store root when --store=local")
 	fs.StringVar(&cfg.s3Bucket, "s3-bucket", "", "S3 bucket when --store=s3")
 	fs.StringVar(&cfg.s3Region, "s3-region", cfg.s3Region, "S3 signing region")
 	fs.StringVar(&cfg.s3Endpoint, "s3-endpoint", "", "S3-compatible endpoint URL")
 	fs.StringVar(&cfg.s3Profile, "s3-profile", "", "AWS shared config profile")
 	fs.BoolVar(&cfg.s3PathStyle, "s3-path-style", cfg.s3PathStyle, "Use path-style S3 addressing")
-	fs.StringVar(&cfg.s3ServerSideEncryption, "s3-sse", "", "Server-side encryption algorithm for uploaded objects, for example AES256 or aws:kms")
-	fs.StringVar(&cfg.s3KMSKeyID, "s3-kms-key-id", "", "KMS key ID when --s3-sse=aws:kms")
+	fs.StringVar(&cfg.s3ServerSideEncryption, "s3-sse", cfg.s3ServerSideEncryption, "Server-side encryption algorithm for uploaded objects: AES256 or aws:kms")
+	fs.StringVar(&cfg.s3KMSKeyID, "s3-kms-key-id", "", "KMS key ARN or bare key ID when --s3-sse=aws:kms; aliases are rejected")
 	fs.BoolVar(&cfg.s3DisableChecksumHeaders, "s3-disable-checksum-headers", false, "Do not send S3 checksum headers; keep metadata and restore-time verification")
 }
 
@@ -199,17 +201,12 @@ func validateStoreFlags(cfg storeFlags) error {
 		if strings.TrimSpace(cfg.s3Bucket) == "" {
 			return errors.New("--s3-bucket is required when --store=s3")
 		}
+		if err := snapshotoffload.ValidateS3StoreEncryption(cfg.s3ServerSideEncryption, cfg.s3KMSKeyID); err != nil {
+			return errors.Wrap(err, "validate s3 encryption")
+		}
 	default:
 		return errors.Errorf("unknown --store %q", cfg.storeKind)
 	}
-	if strings.TrimSpace(cfg.s3KMSKeyID) != "" {
-		if strings.TrimSpace(cfg.s3ServerSideEncryption) == "" {
-			return errors.New("--s3-kms-key-id requires --s3-sse")
-		}
-		if strings.TrimSpace(cfg.s3ServerSideEncryption) != s3SSEAWSKMS {
-			return errors.New("--s3-kms-key-id requires --s3-sse=aws:kms")
-		}
-	}
 	return nil
 }
 

cmd/elastickv-snapshot-offload/main_test.go

@@ -111,7 +111,7 @@ func TestSnapshotOffloadCLIS3KMSRequiresAWSKMS(t *testing.T) {
 		"--data-dir", "data",
 		"--group-id", "1",
 	})
-	require.ErrorContains(t, err, "--s3-kms-key-id requires --s3-sse=aws:kms")
+	require.ErrorContains(t, err, "s3 KMS key id requires aws:kms encryption")
 }
 
 func seedCLISnapshot(t *testing.T, root string, payload []byte, index uint64, term uint64) string {

internal/snapshotoffload/manifest.go

@@ -20,6 +20,7 @@ const (
 var (
 	ErrInvalidOptions = errors.New("snapshot offload: invalid options")
 	ErrIntegrity      = errors.New("snapshot offload: integrity check failed")
+	ErrObjectConflict = errors.New("snapshot offload: object conflict")
 	ErrObjectNotFound = errors.New("snapshot offload: object not found")
 )
 
@@ -82,6 +83,22 @@ func DecodeManifest(data []byte) (Manifest, error) {
 }
 
 func validateManifest(manifest Manifest) error {
+	if err := validateManifestIdentity(manifest); err != nil {
+		return err
+	}
+	if err := validateManifestPayload(manifest.Payload); err != nil {
+		return err
+	}
+	if stringsTrim(manifest.ManifestKey) == "" {
+		return errors.Wrap(ErrInvalidOptions, "manifest key is required")
+	}
+	if len(manifest.ConfState.Voters) == 0 {
+		return errors.Wrap(ErrInvalidOptions, "manifest ConfState requires voters")
+	}
+	return validateManifestConfState(manifest.ConfState)
+}
+
+func validateManifestIdentity(manifest Manifest) error {
 	switch {
 	case manifest.SchemaVersion != ManifestSchemaVersion:
 		return errors.Wrapf(ErrInvalidOptions, "unsupported manifest schema version %d", manifest.SchemaVersion)
@@ -91,18 +108,102 @@ func validateManifest(manifest Manifest) error {
 		return errors.Wrap(ErrInvalidOptions, "snapshot index must be > 0")
 	case manifest.SnapshotTerm == 0:
 		return errors.Wrap(ErrInvalidOptions, "snapshot term must be > 0")
-	case manifest.Payload.Key == "":
+	default:
+		return nil
+	}
+}
+
+func validateManifestPayload(payload PayloadDescriptor) error {
+	switch {
+	case payload.Key == "":
 		return errors.Wrap(ErrInvalidOptions, "payload key is required")
-	case manifest.Payload.Bytes < 0:
+	case payload.Bytes < 0:
 		return errors.Wrap(ErrInvalidOptions, "payload byte count must be >= 0")
-	case !isSHA256Hex(manifest.Payload.SHA256):
+	case !isSHA256Hex(payload.SHA256):
 		return errors.Wrap(ErrInvalidOptions, "payload sha256 must be 64 lowercase hex characters")
-	case manifest.ManifestKey == "":
-		return errors.Wrap(ErrInvalidOptions, "manifest key is required")
+	default:
+		return nil
+	}
+}
+
+func validateManifestConfState(conf ManifestConfState) error {
+	roles := []struct {
+		name   string
+		values []uint64
+	}{
+		{name: "voters", values: conf.Voters},
+		{name: "learners", values: conf.Learners},
+		{name: "voters_outgoing", values: conf.VotersOutgoing},
+		{name: "learners_next", values: conf.LearnersNext},
+	}
+	for _, role := range roles {
+		seen := make(map[uint64]struct{}, len(role.values))
+		for _, id := range role.values {
+			if id == 0 {
+				return errors.Wrapf(ErrInvalidOptions, "conf_state.%s contains zero", role.name)
+			}
+			if _, ok := seen[id]; ok {
+				return errors.Wrapf(ErrInvalidOptions, "conf_state.%s contains duplicate node %d", role.name, id)
+			}
+			seen[id] = struct{}{}
+		}
+	}
+	return validateManifestConfStateMembership(conf)
+}
+
+func validateManifestConfStateMembership(conf ManifestConfState) error {
+	learners := uint64Set(conf.Learners)
+	outgoing := uint64Set(conf.VotersOutgoing)
+	learnersNext := uint64Set(conf.LearnersNext)
+	if err := validateManifestConfStateOverlaps(conf, learners, outgoing, learnersNext); err != nil {
+		return err
+	}
+	for _, id := range conf.LearnersNext {
+		if _, ok := outgoing[id]; !ok {
+			return errors.Wrapf(ErrInvalidOptions, "conf_state.learners_next node %d is not an outgoing voter", id)
+		}
+	}
+	if len(conf.VotersOutgoing) == 0 && conf.AutoLeave {
+		return errors.Wrap(ErrInvalidOptions, "conf_state.auto_leave requires joint consensus")
 	}
 	return nil
 }
 
+func validateManifestConfStateOverlaps(
+	conf ManifestConfState,
+	learners, outgoing, learnersNext map[uint64]struct{},
+) error {
+	for _, id := range conf.Voters {
+		if _, ok := learners[id]; ok {
+			return invalidManifestConfStateOverlap(id, "voters", "learners")
+		}
+		if _, ok := learnersNext[id]; ok {
+			return invalidManifestConfStateOverlap(id, "voters", "learners_next")
+		}
+	}
+	for _, id := range conf.Learners {
+		if _, ok := outgoing[id]; ok {
+			return invalidManifestConfStateOverlap(id, "learners", "voters_outgoing")
+		}
+		if _, ok := learnersNext[id]; ok {
+			return invalidManifestConfStateOverlap(id, "learners", "learners_next")
+		}
+	}
+	return nil
+}
+
+func uint64Set(values []uint64) map[uint64]struct{} {
+	set := make(map[uint64]struct{}, len(values))
+	for _, value := range values {
+		set[value] = struct{}{}
+	}
+	return set
+}
+
+func invalidManifestConfStateOverlap(id uint64, first, second string) error {
+	return errors.Wrapf(ErrInvalidOptions, "node %d appears in conf_state.%s and conf_state.%s", id, first, second)
+}
+
 func verifyManifestSelfHash(manifest Manifest) error {
 	if !isSHA256Hex(manifest.ManifestSHA256) {
 		return errors.Wrap(ErrInvalidOptions, "manifest sha256 must be 64 lowercase hex characters")

internal/snapshotoffload/offload_test.go

@@ -249,6 +249,45 @@ func TestPutManifestReusesExistingManifestAfterCreateConflict(t *testing.T) {
 	require.NotEmpty(t, candidate.ManifestSHA256)
 }
 
+func TestDecodeManifestRejectsInvalidConfStateMembership(t *testing.T) {
+	base := testManifestForValidation(t)
+	testCases := []ManifestConfState{
+		{Voters: nil},
+		{Voters: []uint64{0}},
+		{Voters: []uint64{1, 1}},
+		{Voters: []uint64{1}, Learners: []uint64{2, 2}},
+		{Voters: []uint64{1}, VotersOutgoing: []uint64{2, 2}, LearnersNext: []uint64{2}},
+		{Voters: []uint64{1}, VotersOutgoing: []uint64{2}, LearnersNext: []uint64{2, 2}},
+		{Voters: []uint64{1}, Learners: []uint64{1}},
+		{Voters: []uint64{1}, VotersOutgoing: []uint64{1}, LearnersNext: []uint64{1}},
+		{Voters: []uint64{1}, VotersOutgoing: []uint64{1}, LearnersNext: []uint64{2}},
+		{Voters: []uint64{1}, AutoLeave: true},
+	}
+	for _, confState := range testCases {
+		manifest := base
+		manifest.ConfState = confState
+		raw, _, err := manifest.MarshalCanonical()
+		require.NoError(t, err)
+		_, err = DecodeManifest(raw)
+		require.ErrorIs(t, err, ErrInvalidOptions)
+	}
+}
+
+func TestDecodeManifestAcceptsJointConsensusConfState(t *testing.T) {
+	manifest := testManifestForValidation(t)
+	manifest.ConfState = ManifestConfState{
+		Voters:         []uint64{1, 2},
+		VotersOutgoing: []uint64{1, 3},
+		LearnersNext:   []uint64{3},
+		AutoLeave:      true,
+	}
+	raw, _, err := manifest.MarshalCanonical()
+	require.NoError(t, err)
+	decoded, err := DecodeManifest(raw)
+	require.NoError(t, err)
+	require.Equal(t, manifest.ConfState, decoded.ConfState)
+}
+
 func TestPublishRejectsGroupZeroWithoutSourceClusterBeforePayload(t *testing.T) {
 	ctx := context.Background()
 	root := t.TempDir()
@@ -530,6 +569,30 @@ func TestPrepareRestoreDownloadDirCreatesParentAndCleansOnlyStaleDirs(t *testing
 	require.True(t, os.IsNotExist(err))
 }
 
+func testManifestForValidation(t *testing.T) Manifest {
+	t.Helper()
+	payloadSHA := hexSHA256Bytes([]byte("payload"))
+	key, err := manifestKey("cluster-a", 1, 20, 13)
+	require.NoError(t, err)
+	return Manifest{
+		SchemaVersion: ManifestSchemaVersion,
+		CreatedAt:     time.Unix(400, 0).UTC(),
+		SourceCluster: "cluster-a",
+		GroupID:       1,
+		SnapshotIndex: 20,
+		SnapshotTerm:  13,
+		ConfState: ManifestConfState{
+			Voters: []uint64{1},
+		},
+		Payload: PayloadDescriptor{
+			Key:    "cluster-a/v1/payloads/test.fsm",
+			Bytes:  int64(len("payload")),
+			SHA256: payloadSHA,
+		},
+		ManifestKey: key,
+	}
+}
+
 func seedPhysicalSnapshot(
 	t *testing.T,
 	root string,