laskoviymishka commented on code in PR #1620: URL: https://github.com/apache/iceberg-go/pull/1620#discussion_r3719801410
########## encryption/standard_manager.go: ########## @@ -0,0 +1,511 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package encryption + +import ( + "context" + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "io/fs" + + icebergio "github.com/apache/iceberg-go/io" +) + +// Defaults for [StandardEncryptionManager]. +const ( + // StandardDefaultDEKLength is the default length, in bytes, of the + // per-file data encryption key (DEK) generated for AES-256-GCM. + StandardDefaultDEKLength = 32 + + // StandardDefaultBlockSize is the default plaintext block size, in + // bytes, used to split a file into independently authenticated AES-GCM + // blocks. Blocks allow random access (Seek/ReadAt) without buffering or + // decrypting the whole file. + StandardDefaultBlockSize = 64 * 1024 +) + +// Sentinel errors returned by [StandardEncryptionManager]. +var ( + // ErrKeyIDRequired is returned by + // [StandardEncryptionManager.NewEncryptedOutputFile] when keyID is empty. + // StandardEncryptionManager always encrypts, so it requires a KEK to wrap + // the generated DEK; use [PlaintextEncryptionManager] for unencrypted + // tables instead of passing an empty keyID here. + ErrKeyIDRequired = errors.New("encryption: StandardEncryptionManager requires a non-empty keyID") + + // ErrKeyMetadataRequired is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when keyMetadata is + // empty. StandardEncryptionManager always decrypts, so it requires the + // per-file key metadata produced by [StandardEncryptionManager.NewEncryptedOutputFile]. + ErrKeyMetadataRequired = errors.New("encryption: StandardEncryptionManager requires non-empty key metadata") + + // ErrUnsupportedKeyMetadataVersion is returned when key metadata was + // produced by a newer, incompatible encoding version. + ErrUnsupportedKeyMetadataVersion = errors.New("encryption: unsupported key metadata version") + + // ErrInvalidBlockSize is returned when a configured or decoded block + // size is not positive. + ErrInvalidBlockSize = errors.New("encryption: block size must be positive") + + // ErrInvalidKeyMetadata is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when decoded key + // metadata fails basic sanity checks (e.g. a negative plaintext length + // or a nonce prefix of the wrong size). Key metadata is untrusted input + // on a crypto read path, so it is validated rather than trusted blindly. + ErrInvalidKeyMetadata = errors.New("encryption: invalid key metadata") +) + +// standardKeyMetadataVersion is the current encoding version written by +// [StandardEncryptionManager]. It is bumped whenever the on-disk layout of +// standardKeyMetadata or the block ciphertext format changes incompatibly. +const standardKeyMetadataVersion = 1 + +// standardKeyMetadata is the JSON-encoded structure stored as the opaque +// [EncryptionKeyMetadata] for files produced by [StandardEncryptionManager]. +type standardKeyMetadata struct { + Version int `json:"v"` + KeyID string `json:"key-id"` + WrappedKey []byte `json:"wrapped-key"` + NoncePrefix []byte `json:"nonce-prefix"` + BlockSize int `json:"block-size"` + PlaintextLength int64 `json:"plaintext-length"` +} + +// StandardEncryptionManager is a generic, format-agnostic [EncryptionManager] +// that provides envelope encryption for arbitrary files (e.g. manifests, +// manifest lists, Puffin statistics) using a [KeyManagementClient] to wrap +// and unwrap a fresh AES-256-GCM data encryption key (DEK) per file. +// +// Each file is split into fixed-size plaintext blocks, and each block is +// sealed independently with AES-GCM using a unique nonce (a per-file random +// prefix combined with the block index). This bounds memory usage and +// supports random access (Seek/ReadAt) on the decrypted file without +// buffering or decrypting more than the requested blocks. +// +// StandardEncryptionManager always encrypts and always decrypts: it fails +// closed, returning [ErrKeyIDRequired] or [ErrKeyMetadataRequired] rather +// than silently falling back to plaintext. Use [PlaintextEncryptionManager] +// for tables or files that are not encrypted. +type StandardEncryptionManager struct { + kms KeyManagementClient + dekLength int + blockSize int +} + +var _ EncryptionManager = (*StandardEncryptionManager)(nil) + +// StandardManagerOption configures a [StandardEncryptionManager] created by +// [NewStandardEncryptionManager]. +type StandardManagerOption func(*StandardEncryptionManager) + +// WithDEKLength overrides the default data encryption key length (in bytes). +// Valid AES key lengths are 16, 24, or 32 bytes. +func WithDEKLength(length int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.dekLength = length } +} + +// WithBlockSize overrides the default plaintext block size (in bytes) used +// to split files for independent block-level authentication. +func WithBlockSize(size int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.blockSize = size } +} + +// NewStandardEncryptionManager creates a [StandardEncryptionManager] backed +// by kms. kms must not be nil. +func NewStandardEncryptionManager(kms KeyManagementClient, opts ...StandardManagerOption) *StandardEncryptionManager { + m := &StandardEncryptionManager{ + kms: kms, + dekLength: StandardDefaultDEKLength, + blockSize: StandardDefaultBlockSize, + } + for _, opt := range opts { + opt(m) + } + + return m +} + +// NewEncryptedOutputFile creates a new AES-GCM block-encrypted output file. +// keyID identifies the KEK used to wrap the freshly generated per-file DEK, +// and must be non-empty; otherwise [ErrKeyIDRequired] is returned. +func (m *StandardEncryptionManager) NewEncryptedOutputFile(ctx context.Context, writer icebergio.FileWriter, keyID string) (EncryptedOutputFile, error) { + if keyID == "" { + return nil, ErrKeyIDRequired + } + if m.blockSize <= 0 { + return nil, fmt.Errorf("%w: got %d", ErrInvalidBlockSize, m.blockSize) + } + + var ( + plainDEK, wrappedDEK []byte + err error + ) + if m.kms.SupportsKeyGeneration() { + plainDEK, wrappedDEK, err = m.kms.GenerateKey(ctx, keyID, m.dekLength) + if err != nil { + return nil, fmt.Errorf("encryption: failed to generate DEK: %w", err) + } + } else { + plainDEK = make([]byte, m.dekLength) + if _, err = rand.Read(plainDEK); err != nil { + return nil, fmt.Errorf("encryption: failed to generate DEK: %w", err) + } + if wrappedDEK, err = m.kms.WrapKey(ctx, keyID, plainDEK); err != nil { + return nil, fmt.Errorf("encryption: failed to wrap DEK: %w", err) + } + } + + aead, err := newStandardAEAD(plainDEK) + if err != nil { + return nil, err + } + + noncePrefix := make([]byte, 4) + if _, err := rand.Read(noncePrefix); err != nil { + return nil, fmt.Errorf("encryption: failed to generate nonce prefix: %w", err) + } + + return &standardOutputFile{ + FileWriter: writer, + aead: aead, + noncePrefix: noncePrefix, + blockSize: m.blockSize, + keyID: keyID, + wrappedKey: wrappedDEK, + }, nil +} + Review Comment: agreed with tanmayrauth's call to validate the decoded metadata right after this version check — I'd add one belt-and-suspenders guard on top. Even with the constructor validation, `readBlock` itself does `make([]byte, plainLen+overhead)` with no floor on `plainLen` and no check that `idx` is in `[0, numBlocks())`. It's the actual crypto read path taking numbers derived from untrusted metadata, so I'd have it fail closed there too rather than relying solely on the constructor. Cheap insurance against a future caller that constructs a `standardInputFile` some other way. ########## encryption/standard_manager.go: ########## @@ -0,0 +1,511 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package encryption + +import ( + "context" + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "io/fs" + + icebergio "github.com/apache/iceberg-go/io" +) + +// Defaults for [StandardEncryptionManager]. +const ( + // StandardDefaultDEKLength is the default length, in bytes, of the + // per-file data encryption key (DEK) generated for AES-256-GCM. + StandardDefaultDEKLength = 32 + + // StandardDefaultBlockSize is the default plaintext block size, in + // bytes, used to split a file into independently authenticated AES-GCM + // blocks. Blocks allow random access (Seek/ReadAt) without buffering or + // decrypting the whole file. + StandardDefaultBlockSize = 64 * 1024 +) + +// Sentinel errors returned by [StandardEncryptionManager]. +var ( + // ErrKeyIDRequired is returned by + // [StandardEncryptionManager.NewEncryptedOutputFile] when keyID is empty. + // StandardEncryptionManager always encrypts, so it requires a KEK to wrap + // the generated DEK; use [PlaintextEncryptionManager] for unencrypted + // tables instead of passing an empty keyID here. + ErrKeyIDRequired = errors.New("encryption: StandardEncryptionManager requires a non-empty keyID") + + // ErrKeyMetadataRequired is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when keyMetadata is + // empty. StandardEncryptionManager always decrypts, so it requires the + // per-file key metadata produced by [StandardEncryptionManager.NewEncryptedOutputFile]. + ErrKeyMetadataRequired = errors.New("encryption: StandardEncryptionManager requires non-empty key metadata") + + // ErrUnsupportedKeyMetadataVersion is returned when key metadata was + // produced by a newer, incompatible encoding version. + ErrUnsupportedKeyMetadataVersion = errors.New("encryption: unsupported key metadata version") + + // ErrInvalidBlockSize is returned when a configured or decoded block + // size is not positive. + ErrInvalidBlockSize = errors.New("encryption: block size must be positive") + + // ErrInvalidKeyMetadata is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when decoded key + // metadata fails basic sanity checks (e.g. a negative plaintext length + // or a nonce prefix of the wrong size). Key metadata is untrusted input + // on a crypto read path, so it is validated rather than trusted blindly. + ErrInvalidKeyMetadata = errors.New("encryption: invalid key metadata") +) + +// standardKeyMetadataVersion is the current encoding version written by +// [StandardEncryptionManager]. It is bumped whenever the on-disk layout of +// standardKeyMetadata or the block ciphertext format changes incompatibly. +const standardKeyMetadataVersion = 1 + +// standardKeyMetadata is the JSON-encoded structure stored as the opaque +// [EncryptionKeyMetadata] for files produced by [StandardEncryptionManager]. +type standardKeyMetadata struct { + Version int `json:"v"` + KeyID string `json:"key-id"` + WrappedKey []byte `json:"wrapped-key"` + NoncePrefix []byte `json:"nonce-prefix"` + BlockSize int `json:"block-size"` + PlaintextLength int64 `json:"plaintext-length"` +} + +// StandardEncryptionManager is a generic, format-agnostic [EncryptionManager] +// that provides envelope encryption for arbitrary files (e.g. manifests, +// manifest lists, Puffin statistics) using a [KeyManagementClient] to wrap +// and unwrap a fresh AES-256-GCM data encryption key (DEK) per file. +// +// Each file is split into fixed-size plaintext blocks, and each block is +// sealed independently with AES-GCM using a unique nonce (a per-file random +// prefix combined with the block index). This bounds memory usage and +// supports random access (Seek/ReadAt) on the decrypted file without +// buffering or decrypting more than the requested blocks. +// +// StandardEncryptionManager always encrypts and always decrypts: it fails +// closed, returning [ErrKeyIDRequired] or [ErrKeyMetadataRequired] rather +// than silently falling back to plaintext. Use [PlaintextEncryptionManager] +// for tables or files that are not encrypted. +type StandardEncryptionManager struct { + kms KeyManagementClient + dekLength int + blockSize int +} + +var _ EncryptionManager = (*StandardEncryptionManager)(nil) + +// StandardManagerOption configures a [StandardEncryptionManager] created by +// [NewStandardEncryptionManager]. +type StandardManagerOption func(*StandardEncryptionManager) + +// WithDEKLength overrides the default data encryption key length (in bytes). +// Valid AES key lengths are 16, 24, or 32 bytes. +func WithDEKLength(length int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.dekLength = length } +} + +// WithBlockSize overrides the default plaintext block size (in bytes) used +// to split files for independent block-level authentication. +func WithBlockSize(size int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.blockSize = size } +} + +// NewStandardEncryptionManager creates a [StandardEncryptionManager] backed +// by kms. kms must not be nil. +func NewStandardEncryptionManager(kms KeyManagementClient, opts ...StandardManagerOption) *StandardEncryptionManager { + m := &StandardEncryptionManager{ + kms: kms, + dekLength: StandardDefaultDEKLength, + blockSize: StandardDefaultBlockSize, + } + for _, opt := range opts { + opt(m) + } + + return m +} + +// NewEncryptedOutputFile creates a new AES-GCM block-encrypted output file. +// keyID identifies the KEK used to wrap the freshly generated per-file DEK, +// and must be non-empty; otherwise [ErrKeyIDRequired] is returned. +func (m *StandardEncryptionManager) NewEncryptedOutputFile(ctx context.Context, writer icebergio.FileWriter, keyID string) (EncryptedOutputFile, error) { + if keyID == "" { + return nil, ErrKeyIDRequired + } + if m.blockSize <= 0 { + return nil, fmt.Errorf("%w: got %d", ErrInvalidBlockSize, m.blockSize) + } + + var ( + plainDEK, wrappedDEK []byte + err error + ) + if m.kms.SupportsKeyGeneration() { + plainDEK, wrappedDEK, err = m.kms.GenerateKey(ctx, keyID, m.dekLength) + if err != nil { + return nil, fmt.Errorf("encryption: failed to generate DEK: %w", err) + } + } else { + plainDEK = make([]byte, m.dekLength) Review Comment: This is safe today, but the safety is entirely load-bearing on the DEK being fresh per file, and the 4-byte prefix leaves less room than I'd like. The nonce is 4 random bytes + 8-byte block index, so uniqueness holds only because a new DEK is generated per file — the moment anyone adds DEK caching or reuse, prefix+index collides and GCM nonce reuse is catastrophic. Right now that invariant lives implicitly in the code, not in a comment. I'd either widen the prefix to 8 bytes to buy margin, or at minimum make the invariant explicit at the generation site ("security relies on the DEK being unique per file; do not reuse a DEK across files") so a future change doesn't quietly break it. wdyt? ########## encryption/standard_manager.go: ########## @@ -0,0 +1,511 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package encryption + +import ( + "context" + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "io/fs" + + icebergio "github.com/apache/iceberg-go/io" +) + +// Defaults for [StandardEncryptionManager]. +const ( + // StandardDefaultDEKLength is the default length, in bytes, of the + // per-file data encryption key (DEK) generated for AES-256-GCM. + StandardDefaultDEKLength = 32 + + // StandardDefaultBlockSize is the default plaintext block size, in + // bytes, used to split a file into independently authenticated AES-GCM + // blocks. Blocks allow random access (Seek/ReadAt) without buffering or + // decrypting the whole file. + StandardDefaultBlockSize = 64 * 1024 +) + +// Sentinel errors returned by [StandardEncryptionManager]. +var ( + // ErrKeyIDRequired is returned by + // [StandardEncryptionManager.NewEncryptedOutputFile] when keyID is empty. + // StandardEncryptionManager always encrypts, so it requires a KEK to wrap + // the generated DEK; use [PlaintextEncryptionManager] for unencrypted + // tables instead of passing an empty keyID here. + ErrKeyIDRequired = errors.New("encryption: StandardEncryptionManager requires a non-empty keyID") + + // ErrKeyMetadataRequired is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when keyMetadata is + // empty. StandardEncryptionManager always decrypts, so it requires the + // per-file key metadata produced by [StandardEncryptionManager.NewEncryptedOutputFile]. + ErrKeyMetadataRequired = errors.New("encryption: StandardEncryptionManager requires non-empty key metadata") + + // ErrUnsupportedKeyMetadataVersion is returned when key metadata was + // produced by a newer, incompatible encoding version. + ErrUnsupportedKeyMetadataVersion = errors.New("encryption: unsupported key metadata version") + + // ErrInvalidBlockSize is returned when a configured or decoded block + // size is not positive. + ErrInvalidBlockSize = errors.New("encryption: block size must be positive") + + // ErrInvalidKeyMetadata is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when decoded key + // metadata fails basic sanity checks (e.g. a negative plaintext length + // or a nonce prefix of the wrong size). Key metadata is untrusted input + // on a crypto read path, so it is validated rather than trusted blindly. + ErrInvalidKeyMetadata = errors.New("encryption: invalid key metadata") +) + +// standardKeyMetadataVersion is the current encoding version written by +// [StandardEncryptionManager]. It is bumped whenever the on-disk layout of +// standardKeyMetadata or the block ciphertext format changes incompatibly. +const standardKeyMetadataVersion = 1 + +// standardKeyMetadata is the JSON-encoded structure stored as the opaque +// [EncryptionKeyMetadata] for files produced by [StandardEncryptionManager]. +type standardKeyMetadata struct { + Version int `json:"v"` + KeyID string `json:"key-id"` + WrappedKey []byte `json:"wrapped-key"` + NoncePrefix []byte `json:"nonce-prefix"` + BlockSize int `json:"block-size"` + PlaintextLength int64 `json:"plaintext-length"` +} + +// StandardEncryptionManager is a generic, format-agnostic [EncryptionManager] +// that provides envelope encryption for arbitrary files (e.g. manifests, +// manifest lists, Puffin statistics) using a [KeyManagementClient] to wrap +// and unwrap a fresh AES-256-GCM data encryption key (DEK) per file. +// +// Each file is split into fixed-size plaintext blocks, and each block is +// sealed independently with AES-GCM using a unique nonce (a per-file random +// prefix combined with the block index). This bounds memory usage and +// supports random access (Seek/ReadAt) on the decrypted file without +// buffering or decrypting more than the requested blocks. +// +// StandardEncryptionManager always encrypts and always decrypts: it fails +// closed, returning [ErrKeyIDRequired] or [ErrKeyMetadataRequired] rather +// than silently falling back to plaintext. Use [PlaintextEncryptionManager] +// for tables or files that are not encrypted. +type StandardEncryptionManager struct { + kms KeyManagementClient + dekLength int + blockSize int +} + +var _ EncryptionManager = (*StandardEncryptionManager)(nil) + +// StandardManagerOption configures a [StandardEncryptionManager] created by +// [NewStandardEncryptionManager]. +type StandardManagerOption func(*StandardEncryptionManager) + +// WithDEKLength overrides the default data encryption key length (in bytes). +// Valid AES key lengths are 16, 24, or 32 bytes. +func WithDEKLength(length int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.dekLength = length } +} + +// WithBlockSize overrides the default plaintext block size (in bytes) used +// to split files for independent block-level authentication. +func WithBlockSize(size int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.blockSize = size } +} + +// NewStandardEncryptionManager creates a [StandardEncryptionManager] backed +// by kms. kms must not be nil. Review Comment: agreed with tanmayrauth's blockSize-guard call — the DEK length has the same shape and is the net-new piece I'd add alongside it. `WithDEKLength(17)` is accepted here silently and only blows up much later at `aes.NewCipher` on the first write. Java validates 16/24/32 up front. Since the constructor doesn't return an error, I'd validate both `blockSize` and `dekLength` in `NewEncryptedOutputFile` (which already returns one), or change the constructor to return `(*StandardEncryptionManager, error)`. Either's fine, but I'd rather reject a bad length at construction than at first write. ########## encryption/standard_manager.go: ########## @@ -0,0 +1,511 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package encryption + +import ( + "context" + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "io/fs" + + icebergio "github.com/apache/iceberg-go/io" +) + +// Defaults for [StandardEncryptionManager]. +const ( + // StandardDefaultDEKLength is the default length, in bytes, of the + // per-file data encryption key (DEK) generated for AES-256-GCM. + StandardDefaultDEKLength = 32 + + // StandardDefaultBlockSize is the default plaintext block size, in + // bytes, used to split a file into independently authenticated AES-GCM + // blocks. Blocks allow random access (Seek/ReadAt) without buffering or + // decrypting the whole file. + StandardDefaultBlockSize = 64 * 1024 +) + +// Sentinel errors returned by [StandardEncryptionManager]. +var ( + // ErrKeyIDRequired is returned by + // [StandardEncryptionManager.NewEncryptedOutputFile] when keyID is empty. + // StandardEncryptionManager always encrypts, so it requires a KEK to wrap + // the generated DEK; use [PlaintextEncryptionManager] for unencrypted + // tables instead of passing an empty keyID here. + ErrKeyIDRequired = errors.New("encryption: StandardEncryptionManager requires a non-empty keyID") + + // ErrKeyMetadataRequired is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when keyMetadata is + // empty. StandardEncryptionManager always decrypts, so it requires the + // per-file key metadata produced by [StandardEncryptionManager.NewEncryptedOutputFile]. + ErrKeyMetadataRequired = errors.New("encryption: StandardEncryptionManager requires non-empty key metadata") + + // ErrUnsupportedKeyMetadataVersion is returned when key metadata was + // produced by a newer, incompatible encoding version. + ErrUnsupportedKeyMetadataVersion = errors.New("encryption: unsupported key metadata version") + + // ErrInvalidBlockSize is returned when a configured or decoded block + // size is not positive. + ErrInvalidBlockSize = errors.New("encryption: block size must be positive") + + // ErrInvalidKeyMetadata is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when decoded key + // metadata fails basic sanity checks (e.g. a negative plaintext length + // or a nonce prefix of the wrong size). Key metadata is untrusted input + // on a crypto read path, so it is validated rather than trusted blindly. + ErrInvalidKeyMetadata = errors.New("encryption: invalid key metadata") +) + +// standardKeyMetadataVersion is the current encoding version written by +// [StandardEncryptionManager]. It is bumped whenever the on-disk layout of +// standardKeyMetadata or the block ciphertext format changes incompatibly. +const standardKeyMetadataVersion = 1 + +// standardKeyMetadata is the JSON-encoded structure stored as the opaque +// [EncryptionKeyMetadata] for files produced by [StandardEncryptionManager]. +type standardKeyMetadata struct { + Version int `json:"v"` + KeyID string `json:"key-id"` + WrappedKey []byte `json:"wrapped-key"` + NoncePrefix []byte `json:"nonce-prefix"` + BlockSize int `json:"block-size"` + PlaintextLength int64 `json:"plaintext-length"` +} + +// StandardEncryptionManager is a generic, format-agnostic [EncryptionManager] +// that provides envelope encryption for arbitrary files (e.g. manifests, +// manifest lists, Puffin statistics) using a [KeyManagementClient] to wrap +// and unwrap a fresh AES-256-GCM data encryption key (DEK) per file. +// +// Each file is split into fixed-size plaintext blocks, and each block is +// sealed independently with AES-GCM using a unique nonce (a per-file random +// prefix combined with the block index). This bounds memory usage and +// supports random access (Seek/ReadAt) on the decrypted file without +// buffering or decrypting more than the requested blocks. +// +// StandardEncryptionManager always encrypts and always decrypts: it fails +// closed, returning [ErrKeyIDRequired] or [ErrKeyMetadataRequired] rather +// than silently falling back to plaintext. Use [PlaintextEncryptionManager] +// for tables or files that are not encrypted. +type StandardEncryptionManager struct { + kms KeyManagementClient + dekLength int + blockSize int +} + +var _ EncryptionManager = (*StandardEncryptionManager)(nil) + +// StandardManagerOption configures a [StandardEncryptionManager] created by +// [NewStandardEncryptionManager]. +type StandardManagerOption func(*StandardEncryptionManager) + +// WithDEKLength overrides the default data encryption key length (in bytes). +// Valid AES key lengths are 16, 24, or 32 bytes. +func WithDEKLength(length int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.dekLength = length } +} + +// WithBlockSize overrides the default plaintext block size (in bytes) used +// to split files for independent block-level authentication. +func WithBlockSize(size int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.blockSize = size } +} + +// NewStandardEncryptionManager creates a [StandardEncryptionManager] backed +// by kms. kms must not be nil. +func NewStandardEncryptionManager(kms KeyManagementClient, opts ...StandardManagerOption) *StandardEncryptionManager { + m := &StandardEncryptionManager{ + kms: kms, + dekLength: StandardDefaultDEKLength, + blockSize: StandardDefaultBlockSize, + } + for _, opt := range opts { + opt(m) + } + + return m +} + +// NewEncryptedOutputFile creates a new AES-GCM block-encrypted output file. +// keyID identifies the KEK used to wrap the freshly generated per-file DEK, +// and must be non-empty; otherwise [ErrKeyIDRequired] is returned. +func (m *StandardEncryptionManager) NewEncryptedOutputFile(ctx context.Context, writer icebergio.FileWriter, keyID string) (EncryptedOutputFile, error) { + if keyID == "" { + return nil, ErrKeyIDRequired + } + if m.blockSize <= 0 { + return nil, fmt.Errorf("%w: got %d", ErrInvalidBlockSize, m.blockSize) + } + + var ( + plainDEK, wrappedDEK []byte + err error + ) + if m.kms.SupportsKeyGeneration() { + plainDEK, wrappedDEK, err = m.kms.GenerateKey(ctx, keyID, m.dekLength) + if err != nil { + return nil, fmt.Errorf("encryption: failed to generate DEK: %w", err) + } + } else { + plainDEK = make([]byte, m.dekLength) + if _, err = rand.Read(plainDEK); err != nil { + return nil, fmt.Errorf("encryption: failed to generate DEK: %w", err) + } + if wrappedDEK, err = m.kms.WrapKey(ctx, keyID, plainDEK); err != nil { + return nil, fmt.Errorf("encryption: failed to wrap DEK: %w", err) + } + } + + aead, err := newStandardAEAD(plainDEK) + if err != nil { + return nil, err + } + + noncePrefix := make([]byte, 4) + if _, err := rand.Read(noncePrefix); err != nil { + return nil, fmt.Errorf("encryption: failed to generate nonce prefix: %w", err) + } + + return &standardOutputFile{ + FileWriter: writer, + aead: aead, + noncePrefix: noncePrefix, + blockSize: m.blockSize, + keyID: keyID, + wrappedKey: wrappedDEK, + }, nil +} + +// NewDecryptedInputFile wraps file for transparent block-level AES-GCM +// decryption. keyMetadata must be the non-empty blob produced by +// [StandardEncryptionManager.NewEncryptedOutputFile]; otherwise +// [ErrKeyMetadataRequired] is returned. +func (m *StandardEncryptionManager) NewDecryptedInputFile(ctx context.Context, file icebergio.File, keyMetadata EncryptionKeyMetadata) (EncryptedInputFile, error) { + if len(keyMetadata) == 0 { + return nil, ErrKeyMetadataRequired + } + + var meta standardKeyMetadata + if err := json.Unmarshal(keyMetadata, &meta); err != nil { + return nil, fmt.Errorf("encryption: failed to decode key metadata: %w", err) + } + if meta.Version != standardKeyMetadataVersion { + return nil, fmt.Errorf("%w: %d", ErrUnsupportedKeyMetadataVersion, meta.Version) + } + if meta.BlockSize <= 0 { + return nil, fmt.Errorf("%w: block-size must be positive, got %d", ErrInvalidKeyMetadata, meta.BlockSize) + } + if meta.PlaintextLength < 0 { + return nil, fmt.Errorf("%w: plaintext-length must be non-negative, got %d", ErrInvalidKeyMetadata, meta.PlaintextLength) + } + if len(meta.NoncePrefix) != 4 { + return nil, fmt.Errorf("%w: nonce-prefix must be 4 bytes, got %d", ErrInvalidKeyMetadata, len(meta.NoncePrefix)) + } + + plainDEK, err := m.kms.UnwrapKey(ctx, meta.KeyID, meta.WrappedKey) + if err != nil { + return nil, fmt.Errorf("encryption: failed to unwrap DEK: %w", err) + } + + aead, err := newStandardAEAD(plainDEK) + if err != nil { + return nil, err + } + + return &standardInputFile{ + underlying: file, + aead: aead, + noncePrefix: meta.NoncePrefix, + blockSize: meta.BlockSize, + plaintextLength: meta.PlaintextLength, + keyMetadata: keyMetadata, + }, nil +} + +func newStandardAEAD(key []byte) (cipher.AEAD, error) { + block, err := aes.NewCipher(key) + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrInvalidKeyLength, err) + } + gcm, err := cipher.NewGCM(block) + if err != nil { + return nil, fmt.Errorf("encryption: failed to create GCM: %w", err) + } + + return gcm, nil +} + +// standardBlockNonce derives the AES-GCM nonce for blockIndex: the 4-byte +// per-file random prefix followed by the 8-byte big-endian block index. The +// (key, nonce) pair is unique per block since the DEK is fresh per file and +// no two blocks in the same file share an index. +func standardBlockNonce(prefix []byte, blockIndex uint64) []byte { + nonce := make([]byte, 12) + copy(nonce, prefix) + binary.BigEndian.PutUint64(nonce[4:], blockIndex) + + return nonce +} + +// standardOutputFile is an [EncryptedOutputFile] that seals fixed-size +// plaintext blocks with AES-GCM as they are written. +type standardOutputFile struct { + icebergio.FileWriter + + aead cipher.AEAD + noncePrefix []byte + blockSize int + keyID string + wrappedKey []byte + + buf []byte + blockIndex uint64 + written int64 + closed bool + + keyMetadata EncryptionKeyMetadata +} + +var _ EncryptedOutputFile = (*standardOutputFile)(nil) + +func (f *standardOutputFile) Write(p []byte) (int, error) { + total := len(p) + for len(p) > 0 { + space := f.blockSize - len(f.buf) + n := min(space, len(p)) + f.buf = append(f.buf, p[:n]...) + p = p[n:] + f.written += int64(n) + if len(f.buf) == f.blockSize { + if err := f.flushBlock(); err != nil { + return total - len(p), err + } + } + } + + return total, nil +} + +func (f *standardOutputFile) flushBlock() error { + ciphertext := f.aead.Seal(nil, standardBlockNonce(f.noncePrefix, f.blockIndex), f.buf, nil) + if _, err := f.FileWriter.Write(ciphertext); err != nil { + return fmt.Errorf("encryption: failed to write encrypted block: %w", err) + } + f.blockIndex++ + f.buf = f.buf[:0] + + return nil +} + +// ReadFrom copies from r, encrypting as data is written, satisfying +// io.ReaderFrom (required by [icebergio.FileWriter]). +func (f *standardOutputFile) ReadFrom(r io.Reader) (int64, error) { + buf := make([]byte, 32*1024) + var total int64 + for { + n, err := r.Read(buf) Review Comment: `closed` gets set before the final flush, so a `Close` that fails its flush can't be retried. The second `Close` hits the `if f.closed { return nil }` guard and returns nil without re-flushing, so the caller sees success on a file that never finished writing (and `KeyMetadata` stays empty). The flush-failure path also drops the `FileWriter.Close()` error via `_ =`. I'd set `closed = true` only on the success path (or track a sticky `f.err` and return it on every subsequent `Close`). A `failWriter` stub that errors on the Nth block would let us assert `Close` errors, `KeyMetadata()` is nil, and a repeat `Close` still errors. ########## encryption/standard_manager.go: ########## @@ -0,0 +1,511 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package encryption + +import ( + "context" + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "io/fs" + + icebergio "github.com/apache/iceberg-go/io" +) + +// Defaults for [StandardEncryptionManager]. +const ( + // StandardDefaultDEKLength is the default length, in bytes, of the + // per-file data encryption key (DEK) generated for AES-256-GCM. + StandardDefaultDEKLength = 32 + + // StandardDefaultBlockSize is the default plaintext block size, in + // bytes, used to split a file into independently authenticated AES-GCM + // blocks. Blocks allow random access (Seek/ReadAt) without buffering or + // decrypting the whole file. + StandardDefaultBlockSize = 64 * 1024 +) + +// Sentinel errors returned by [StandardEncryptionManager]. +var ( + // ErrKeyIDRequired is returned by + // [StandardEncryptionManager.NewEncryptedOutputFile] when keyID is empty. + // StandardEncryptionManager always encrypts, so it requires a KEK to wrap + // the generated DEK; use [PlaintextEncryptionManager] for unencrypted + // tables instead of passing an empty keyID here. + ErrKeyIDRequired = errors.New("encryption: StandardEncryptionManager requires a non-empty keyID") + + // ErrKeyMetadataRequired is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when keyMetadata is + // empty. StandardEncryptionManager always decrypts, so it requires the + // per-file key metadata produced by [StandardEncryptionManager.NewEncryptedOutputFile]. + ErrKeyMetadataRequired = errors.New("encryption: StandardEncryptionManager requires non-empty key metadata") + + // ErrUnsupportedKeyMetadataVersion is returned when key metadata was + // produced by a newer, incompatible encoding version. + ErrUnsupportedKeyMetadataVersion = errors.New("encryption: unsupported key metadata version") + + // ErrInvalidBlockSize is returned when a configured or decoded block + // size is not positive. + ErrInvalidBlockSize = errors.New("encryption: block size must be positive") + + // ErrInvalidKeyMetadata is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when decoded key + // metadata fails basic sanity checks (e.g. a negative plaintext length + // or a nonce prefix of the wrong size). Key metadata is untrusted input + // on a crypto read path, so it is validated rather than trusted blindly. + ErrInvalidKeyMetadata = errors.New("encryption: invalid key metadata") +) + +// standardKeyMetadataVersion is the current encoding version written by +// [StandardEncryptionManager]. It is bumped whenever the on-disk layout of +// standardKeyMetadata or the block ciphertext format changes incompatibly. +const standardKeyMetadataVersion = 1 + +// standardKeyMetadata is the JSON-encoded structure stored as the opaque +// [EncryptionKeyMetadata] for files produced by [StandardEncryptionManager]. +type standardKeyMetadata struct { + Version int `json:"v"` + KeyID string `json:"key-id"` + WrappedKey []byte `json:"wrapped-key"` + NoncePrefix []byte `json:"nonce-prefix"` + BlockSize int `json:"block-size"` + PlaintextLength int64 `json:"plaintext-length"` +} + +// StandardEncryptionManager is a generic, format-agnostic [EncryptionManager] +// that provides envelope encryption for arbitrary files (e.g. manifests, +// manifest lists, Puffin statistics) using a [KeyManagementClient] to wrap +// and unwrap a fresh AES-256-GCM data encryption key (DEK) per file. +// +// Each file is split into fixed-size plaintext blocks, and each block is +// sealed independently with AES-GCM using a unique nonce (a per-file random +// prefix combined with the block index). This bounds memory usage and +// supports random access (Seek/ReadAt) on the decrypted file without +// buffering or decrypting more than the requested blocks. +// +// StandardEncryptionManager always encrypts and always decrypts: it fails +// closed, returning [ErrKeyIDRequired] or [ErrKeyMetadataRequired] rather +// than silently falling back to plaintext. Use [PlaintextEncryptionManager] +// for tables or files that are not encrypted. +type StandardEncryptionManager struct { + kms KeyManagementClient + dekLength int + blockSize int +} + +var _ EncryptionManager = (*StandardEncryptionManager)(nil) + +// StandardManagerOption configures a [StandardEncryptionManager] created by +// [NewStandardEncryptionManager]. +type StandardManagerOption func(*StandardEncryptionManager) + +// WithDEKLength overrides the default data encryption key length (in bytes). +// Valid AES key lengths are 16, 24, or 32 bytes. +func WithDEKLength(length int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.dekLength = length } +} + +// WithBlockSize overrides the default plaintext block size (in bytes) used +// to split files for independent block-level authentication. +func WithBlockSize(size int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.blockSize = size } +} + +// NewStandardEncryptionManager creates a [StandardEncryptionManager] backed +// by kms. kms must not be nil. +func NewStandardEncryptionManager(kms KeyManagementClient, opts ...StandardManagerOption) *StandardEncryptionManager { + m := &StandardEncryptionManager{ + kms: kms, + dekLength: StandardDefaultDEKLength, + blockSize: StandardDefaultBlockSize, + } + for _, opt := range opts { + opt(m) + } + + return m +} + +// NewEncryptedOutputFile creates a new AES-GCM block-encrypted output file. +// keyID identifies the KEK used to wrap the freshly generated per-file DEK, +// and must be non-empty; otherwise [ErrKeyIDRequired] is returned. +func (m *StandardEncryptionManager) NewEncryptedOutputFile(ctx context.Context, writer icebergio.FileWriter, keyID string) (EncryptedOutputFile, error) { + if keyID == "" { + return nil, ErrKeyIDRequired + } + if m.blockSize <= 0 { + return nil, fmt.Errorf("%w: got %d", ErrInvalidBlockSize, m.blockSize) + } + + var ( + plainDEK, wrappedDEK []byte + err error + ) + if m.kms.SupportsKeyGeneration() { + plainDEK, wrappedDEK, err = m.kms.GenerateKey(ctx, keyID, m.dekLength) + if err != nil { + return nil, fmt.Errorf("encryption: failed to generate DEK: %w", err) + } + } else { + plainDEK = make([]byte, m.dekLength) + if _, err = rand.Read(plainDEK); err != nil { + return nil, fmt.Errorf("encryption: failed to generate DEK: %w", err) + } + if wrappedDEK, err = m.kms.WrapKey(ctx, keyID, plainDEK); err != nil { + return nil, fmt.Errorf("encryption: failed to wrap DEK: %w", err) + } + } + + aead, err := newStandardAEAD(plainDEK) + if err != nil { + return nil, err + } + + noncePrefix := make([]byte, 4) + if _, err := rand.Read(noncePrefix); err != nil { + return nil, fmt.Errorf("encryption: failed to generate nonce prefix: %w", err) + } + + return &standardOutputFile{ + FileWriter: writer, + aead: aead, + noncePrefix: noncePrefix, + blockSize: m.blockSize, + keyID: keyID, + wrappedKey: wrappedDEK, + }, nil +} + +// NewDecryptedInputFile wraps file for transparent block-level AES-GCM +// decryption. keyMetadata must be the non-empty blob produced by +// [StandardEncryptionManager.NewEncryptedOutputFile]; otherwise +// [ErrKeyMetadataRequired] is returned. +func (m *StandardEncryptionManager) NewDecryptedInputFile(ctx context.Context, file icebergio.File, keyMetadata EncryptionKeyMetadata) (EncryptedInputFile, error) { + if len(keyMetadata) == 0 { + return nil, ErrKeyMetadataRequired + } + + var meta standardKeyMetadata + if err := json.Unmarshal(keyMetadata, &meta); err != nil { + return nil, fmt.Errorf("encryption: failed to decode key metadata: %w", err) + } + if meta.Version != standardKeyMetadataVersion { + return nil, fmt.Errorf("%w: %d", ErrUnsupportedKeyMetadataVersion, meta.Version) + } + if meta.BlockSize <= 0 { + return nil, fmt.Errorf("%w: block-size must be positive, got %d", ErrInvalidKeyMetadata, meta.BlockSize) + } + if meta.PlaintextLength < 0 { + return nil, fmt.Errorf("%w: plaintext-length must be non-negative, got %d", ErrInvalidKeyMetadata, meta.PlaintextLength) + } + if len(meta.NoncePrefix) != 4 { + return nil, fmt.Errorf("%w: nonce-prefix must be 4 bytes, got %d", ErrInvalidKeyMetadata, len(meta.NoncePrefix)) + } + + plainDEK, err := m.kms.UnwrapKey(ctx, meta.KeyID, meta.WrappedKey) + if err != nil { + return nil, fmt.Errorf("encryption: failed to unwrap DEK: %w", err) + } + + aead, err := newStandardAEAD(plainDEK) + if err != nil { + return nil, err + } + + return &standardInputFile{ + underlying: file, + aead: aead, + noncePrefix: meta.NoncePrefix, + blockSize: meta.BlockSize, + plaintextLength: meta.PlaintextLength, + keyMetadata: keyMetadata, + }, nil +} + +func newStandardAEAD(key []byte) (cipher.AEAD, error) { + block, err := aes.NewCipher(key) + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrInvalidKeyLength, err) + } + gcm, err := cipher.NewGCM(block) + if err != nil { + return nil, fmt.Errorf("encryption: failed to create GCM: %w", err) + } + + return gcm, nil +} + +// standardBlockNonce derives the AES-GCM nonce for blockIndex: the 4-byte +// per-file random prefix followed by the 8-byte big-endian block index. The +// (key, nonce) pair is unique per block since the DEK is fresh per file and +// no two blocks in the same file share an index. +func standardBlockNonce(prefix []byte, blockIndex uint64) []byte { + nonce := make([]byte, 12) + copy(nonce, prefix) + binary.BigEndian.PutUint64(nonce[4:], blockIndex) + + return nonce +} + +// standardOutputFile is an [EncryptedOutputFile] that seals fixed-size +// plaintext blocks with AES-GCM as they are written. +type standardOutputFile struct { + icebergio.FileWriter + + aead cipher.AEAD + noncePrefix []byte + blockSize int + keyID string + wrappedKey []byte + + buf []byte + blockIndex uint64 + written int64 + closed bool + + keyMetadata EncryptionKeyMetadata +} Review Comment: just adding a concrete data point to tanmayrauth's #1289 interop flag, not asking for anything here. For whoever picks up that discussion, here's the exact byte-level gap vs Java's `AesGcmOutputStream`: no 8-byte "AGS1" + block-size header, block layout is `ciphertext‖tag` with the nonce *derived* rather than Java's inline random 12-byte nonce, AAD is nil vs Java's `fileAadPrefix‖ordinal`, 64 KiB blocks vs 1 MiB, and `key_metadata` is our JSON struct vs Java's Avro `StandardKeyMetadata`. So the formats are byte-incompatible in both directions today. Two things that lean toward resolving it sooner rather than later: the type reuses the name `StandardEncryptionManager`, which is the canonical cross-engine impl elsewhere, so a reader reasonably assumes interop and gets silently-unreadable tables; and only the DEK is wrapped here — the nonce prefix, block size, and plaintext length ride in cleartext JSON, where Java envelope-encrypts the whole blob. Both are inputs for #1289, not changes for this PR. ########## encryption/standard_manager.go: ########## @@ -0,0 +1,511 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package encryption + +import ( + "context" + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "io/fs" + + icebergio "github.com/apache/iceberg-go/io" +) + +// Defaults for [StandardEncryptionManager]. +const ( + // StandardDefaultDEKLength is the default length, in bytes, of the + // per-file data encryption key (DEK) generated for AES-256-GCM. + StandardDefaultDEKLength = 32 + + // StandardDefaultBlockSize is the default plaintext block size, in + // bytes, used to split a file into independently authenticated AES-GCM + // blocks. Blocks allow random access (Seek/ReadAt) without buffering or + // decrypting the whole file. + StandardDefaultBlockSize = 64 * 1024 +) + +// Sentinel errors returned by [StandardEncryptionManager]. +var ( + // ErrKeyIDRequired is returned by + // [StandardEncryptionManager.NewEncryptedOutputFile] when keyID is empty. + // StandardEncryptionManager always encrypts, so it requires a KEK to wrap + // the generated DEK; use [PlaintextEncryptionManager] for unencrypted + // tables instead of passing an empty keyID here. + ErrKeyIDRequired = errors.New("encryption: StandardEncryptionManager requires a non-empty keyID") + + // ErrKeyMetadataRequired is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when keyMetadata is + // empty. StandardEncryptionManager always decrypts, so it requires the + // per-file key metadata produced by [StandardEncryptionManager.NewEncryptedOutputFile]. + ErrKeyMetadataRequired = errors.New("encryption: StandardEncryptionManager requires non-empty key metadata") + + // ErrUnsupportedKeyMetadataVersion is returned when key metadata was + // produced by a newer, incompatible encoding version. + ErrUnsupportedKeyMetadataVersion = errors.New("encryption: unsupported key metadata version") + + // ErrInvalidBlockSize is returned when a configured or decoded block + // size is not positive. + ErrInvalidBlockSize = errors.New("encryption: block size must be positive") + + // ErrInvalidKeyMetadata is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when decoded key + // metadata fails basic sanity checks (e.g. a negative plaintext length + // or a nonce prefix of the wrong size). Key metadata is untrusted input + // on a crypto read path, so it is validated rather than trusted blindly. + ErrInvalidKeyMetadata = errors.New("encryption: invalid key metadata") +) + +// standardKeyMetadataVersion is the current encoding version written by +// [StandardEncryptionManager]. It is bumped whenever the on-disk layout of +// standardKeyMetadata or the block ciphertext format changes incompatibly. +const standardKeyMetadataVersion = 1 + +// standardKeyMetadata is the JSON-encoded structure stored as the opaque +// [EncryptionKeyMetadata] for files produced by [StandardEncryptionManager]. +type standardKeyMetadata struct { + Version int `json:"v"` + KeyID string `json:"key-id"` + WrappedKey []byte `json:"wrapped-key"` + NoncePrefix []byte `json:"nonce-prefix"` + BlockSize int `json:"block-size"` + PlaintextLength int64 `json:"plaintext-length"` +} + +// StandardEncryptionManager is a generic, format-agnostic [EncryptionManager] +// that provides envelope encryption for arbitrary files (e.g. manifests, +// manifest lists, Puffin statistics) using a [KeyManagementClient] to wrap +// and unwrap a fresh AES-256-GCM data encryption key (DEK) per file. +// +// Each file is split into fixed-size plaintext blocks, and each block is +// sealed independently with AES-GCM using a unique nonce (a per-file random +// prefix combined with the block index). This bounds memory usage and +// supports random access (Seek/ReadAt) on the decrypted file without +// buffering or decrypting more than the requested blocks. +// +// StandardEncryptionManager always encrypts and always decrypts: it fails +// closed, returning [ErrKeyIDRequired] or [ErrKeyMetadataRequired] rather +// than silently falling back to plaintext. Use [PlaintextEncryptionManager] +// for tables or files that are not encrypted. +type StandardEncryptionManager struct { + kms KeyManagementClient + dekLength int + blockSize int +} + +var _ EncryptionManager = (*StandardEncryptionManager)(nil) + +// StandardManagerOption configures a [StandardEncryptionManager] created by +// [NewStandardEncryptionManager]. +type StandardManagerOption func(*StandardEncryptionManager) + +// WithDEKLength overrides the default data encryption key length (in bytes). +// Valid AES key lengths are 16, 24, or 32 bytes. +func WithDEKLength(length int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.dekLength = length } +} + +// WithBlockSize overrides the default plaintext block size (in bytes) used +// to split files for independent block-level authentication. +func WithBlockSize(size int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.blockSize = size } +} + +// NewStandardEncryptionManager creates a [StandardEncryptionManager] backed +// by kms. kms must not be nil. +func NewStandardEncryptionManager(kms KeyManagementClient, opts ...StandardManagerOption) *StandardEncryptionManager { + m := &StandardEncryptionManager{ + kms: kms, + dekLength: StandardDefaultDEKLength, + blockSize: StandardDefaultBlockSize, + } + for _, opt := range opts { + opt(m) + } + + return m +} + +// NewEncryptedOutputFile creates a new AES-GCM block-encrypted output file. +// keyID identifies the KEK used to wrap the freshly generated per-file DEK, +// and must be non-empty; otherwise [ErrKeyIDRequired] is returned. +func (m *StandardEncryptionManager) NewEncryptedOutputFile(ctx context.Context, writer icebergio.FileWriter, keyID string) (EncryptedOutputFile, error) { + if keyID == "" { + return nil, ErrKeyIDRequired + } + if m.blockSize <= 0 { + return nil, fmt.Errorf("%w: got %d", ErrInvalidBlockSize, m.blockSize) + } + + var ( + plainDEK, wrappedDEK []byte + err error + ) + if m.kms.SupportsKeyGeneration() { + plainDEK, wrappedDEK, err = m.kms.GenerateKey(ctx, keyID, m.dekLength) + if err != nil { + return nil, fmt.Errorf("encryption: failed to generate DEK: %w", err) + } + } else { + plainDEK = make([]byte, m.dekLength) + if _, err = rand.Read(plainDEK); err != nil { + return nil, fmt.Errorf("encryption: failed to generate DEK: %w", err) + } + if wrappedDEK, err = m.kms.WrapKey(ctx, keyID, plainDEK); err != nil { + return nil, fmt.Errorf("encryption: failed to wrap DEK: %w", err) + } + } + + aead, err := newStandardAEAD(plainDEK) + if err != nil { + return nil, err + } + + noncePrefix := make([]byte, 4) + if _, err := rand.Read(noncePrefix); err != nil { + return nil, fmt.Errorf("encryption: failed to generate nonce prefix: %w", err) + } + + return &standardOutputFile{ + FileWriter: writer, + aead: aead, + noncePrefix: noncePrefix, + blockSize: m.blockSize, + keyID: keyID, + wrappedKey: wrappedDEK, + }, nil +} + +// NewDecryptedInputFile wraps file for transparent block-level AES-GCM +// decryption. keyMetadata must be the non-empty blob produced by +// [StandardEncryptionManager.NewEncryptedOutputFile]; otherwise +// [ErrKeyMetadataRequired] is returned. +func (m *StandardEncryptionManager) NewDecryptedInputFile(ctx context.Context, file icebergio.File, keyMetadata EncryptionKeyMetadata) (EncryptedInputFile, error) { + if len(keyMetadata) == 0 { + return nil, ErrKeyMetadataRequired + } + + var meta standardKeyMetadata + if err := json.Unmarshal(keyMetadata, &meta); err != nil { + return nil, fmt.Errorf("encryption: failed to decode key metadata: %w", err) + } + if meta.Version != standardKeyMetadataVersion { + return nil, fmt.Errorf("%w: %d", ErrUnsupportedKeyMetadataVersion, meta.Version) + } + if meta.BlockSize <= 0 { + return nil, fmt.Errorf("%w: block-size must be positive, got %d", ErrInvalidKeyMetadata, meta.BlockSize) + } + if meta.PlaintextLength < 0 { + return nil, fmt.Errorf("%w: plaintext-length must be non-negative, got %d", ErrInvalidKeyMetadata, meta.PlaintextLength) + } + if len(meta.NoncePrefix) != 4 { + return nil, fmt.Errorf("%w: nonce-prefix must be 4 bytes, got %d", ErrInvalidKeyMetadata, len(meta.NoncePrefix)) + } + + plainDEK, err := m.kms.UnwrapKey(ctx, meta.KeyID, meta.WrappedKey) + if err != nil { + return nil, fmt.Errorf("encryption: failed to unwrap DEK: %w", err) + } + + aead, err := newStandardAEAD(plainDEK) + if err != nil { + return nil, err + } + + return &standardInputFile{ + underlying: file, + aead: aead, + noncePrefix: meta.NoncePrefix, + blockSize: meta.BlockSize, + plaintextLength: meta.PlaintextLength, + keyMetadata: keyMetadata, + }, nil +} + +func newStandardAEAD(key []byte) (cipher.AEAD, error) { + block, err := aes.NewCipher(key) + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrInvalidKeyLength, err) + } + gcm, err := cipher.NewGCM(block) + if err != nil { + return nil, fmt.Errorf("encryption: failed to create GCM: %w", err) + } + + return gcm, nil +} + +// standardBlockNonce derives the AES-GCM nonce for blockIndex: the 4-byte +// per-file random prefix followed by the 8-byte big-endian block index. The +// (key, nonce) pair is unique per block since the DEK is fresh per file and +// no two blocks in the same file share an index. +func standardBlockNonce(prefix []byte, blockIndex uint64) []byte { + nonce := make([]byte, 12) + copy(nonce, prefix) + binary.BigEndian.PutUint64(nonce[4:], blockIndex) + + return nonce +} + +// standardOutputFile is an [EncryptedOutputFile] that seals fixed-size +// plaintext blocks with AES-GCM as they are written. +type standardOutputFile struct { + icebergio.FileWriter + + aead cipher.AEAD + noncePrefix []byte + blockSize int + keyID string + wrappedKey []byte + + buf []byte + blockIndex uint64 + written int64 + closed bool + + keyMetadata EncryptionKeyMetadata +} + +var _ EncryptedOutputFile = (*standardOutputFile)(nil) + +func (f *standardOutputFile) Write(p []byte) (int, error) { + total := len(p) + for len(p) > 0 { + space := f.blockSize - len(f.buf) + n := min(space, len(p)) + f.buf = append(f.buf, p[:n]...) + p = p[n:] + f.written += int64(n) + if len(f.buf) == f.blockSize { + if err := f.flushBlock(); err != nil { + return total - len(p), err + } + } + } + + return total, nil +} + +func (f *standardOutputFile) flushBlock() error { + ciphertext := f.aead.Seal(nil, standardBlockNonce(f.noncePrefix, f.blockIndex), f.buf, nil) + if _, err := f.FileWriter.Write(ciphertext); err != nil { + return fmt.Errorf("encryption: failed to write encrypted block: %w", err) + } + f.blockIndex++ + f.buf = f.buf[:0] + + return nil +} + +// ReadFrom copies from r, encrypting as data is written, satisfying +// io.ReaderFrom (required by [icebergio.FileWriter]). +func (f *standardOutputFile) ReadFrom(r io.Reader) (int64, error) { + buf := make([]byte, 32*1024) + var total int64 + for { + n, err := r.Read(buf) + if n > 0 { + wn, werr := f.Write(buf[:n]) + total += int64(wn) + if werr != nil { + return total, werr + } + } + if err == io.EOF { + break + } + if err != nil { + return total, err + } + } + + return total, nil +} + +func (f *standardOutputFile) Close() error { + if f.closed { + return nil + } + f.closed = true + + if len(f.buf) > 0 { + if err := f.flushBlock(); err != nil { + _ = f.FileWriter.Close() + + return err + } + } + + meta := standardKeyMetadata{ + Version: standardKeyMetadataVersion, + KeyID: f.keyID, + WrappedKey: f.wrappedKey, + NoncePrefix: f.noncePrefix, + BlockSize: f.blockSize, + PlaintextLength: f.written, + } + encoded, err := json.Marshal(meta) + if err != nil { + _ = f.FileWriter.Close() + + return fmt.Errorf("encryption: failed to encode key metadata: %w", err) + } + f.keyMetadata = encoded + + return f.FileWriter.Close() +} + +// KeyMetadata returns the finalized per-file key metadata. It is only +// populated after Close has been called. +func (f *standardOutputFile) KeyMetadata() EncryptionKeyMetadata { return f.keyMetadata } + +// standardInputFile is an [EncryptedInputFile] that decrypts fixed-size +// AES-GCM blocks on demand, supporting random access via ReadAt/Seek. +type standardInputFile struct { + underlying icebergio.File + aead cipher.AEAD + noncePrefix []byte + blockSize int + plaintextLength int64 + keyMetadata EncryptionKeyMetadata + + pos int64 +} + +var _ EncryptedInputFile = (*standardInputFile)(nil) + +func (f *standardInputFile) numBlocks() int64 { Review Comment: This discards a short read, and I think it breaks the last block of every file whose size isn't a block multiple on a real backend. On S3 or the local fs, `ReadAt` fills a partial final block and returns `(n, io.EOF)` with `n < len(ciphertext)`. We drop the EOF here, so `aead.Open` gets a buffer that's zero-padded past `n` and fails auth — which then surfaces as `ErrAuthenticationFailed`, not truncation. The in-memory test file never hits this because `bytes.Reader` fills the whole slice, so the suite is green while real reads fail. I'd capture `n` and slice `ciphertext = ciphertext[:n]` before `Open`, and treat a genuinely short *non-final* block as a distinct truncation error rather than letting it fall through to auth-failed. Worth a test backed by a reader that returns `(n, io.EOF)` so we actually catch this class. ########## encryption/standard_manager.go: ########## @@ -0,0 +1,511 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package encryption + +import ( + "context" + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "io/fs" + + icebergio "github.com/apache/iceberg-go/io" +) + +// Defaults for [StandardEncryptionManager]. +const ( + // StandardDefaultDEKLength is the default length, in bytes, of the + // per-file data encryption key (DEK) generated for AES-256-GCM. + StandardDefaultDEKLength = 32 + + // StandardDefaultBlockSize is the default plaintext block size, in + // bytes, used to split a file into independently authenticated AES-GCM + // blocks. Blocks allow random access (Seek/ReadAt) without buffering or + // decrypting the whole file. + StandardDefaultBlockSize = 64 * 1024 +) + +// Sentinel errors returned by [StandardEncryptionManager]. +var ( + // ErrKeyIDRequired is returned by + // [StandardEncryptionManager.NewEncryptedOutputFile] when keyID is empty. + // StandardEncryptionManager always encrypts, so it requires a KEK to wrap + // the generated DEK; use [PlaintextEncryptionManager] for unencrypted + // tables instead of passing an empty keyID here. + ErrKeyIDRequired = errors.New("encryption: StandardEncryptionManager requires a non-empty keyID") + + // ErrKeyMetadataRequired is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when keyMetadata is + // empty. StandardEncryptionManager always decrypts, so it requires the + // per-file key metadata produced by [StandardEncryptionManager.NewEncryptedOutputFile]. + ErrKeyMetadataRequired = errors.New("encryption: StandardEncryptionManager requires non-empty key metadata") + + // ErrUnsupportedKeyMetadataVersion is returned when key metadata was + // produced by a newer, incompatible encoding version. + ErrUnsupportedKeyMetadataVersion = errors.New("encryption: unsupported key metadata version") + + // ErrInvalidBlockSize is returned when a configured or decoded block + // size is not positive. + ErrInvalidBlockSize = errors.New("encryption: block size must be positive") + + // ErrInvalidKeyMetadata is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when decoded key + // metadata fails basic sanity checks (e.g. a negative plaintext length + // or a nonce prefix of the wrong size). Key metadata is untrusted input + // on a crypto read path, so it is validated rather than trusted blindly. + ErrInvalidKeyMetadata = errors.New("encryption: invalid key metadata") +) + +// standardKeyMetadataVersion is the current encoding version written by +// [StandardEncryptionManager]. It is bumped whenever the on-disk layout of +// standardKeyMetadata or the block ciphertext format changes incompatibly. +const standardKeyMetadataVersion = 1 + +// standardKeyMetadata is the JSON-encoded structure stored as the opaque +// [EncryptionKeyMetadata] for files produced by [StandardEncryptionManager]. +type standardKeyMetadata struct { + Version int `json:"v"` + KeyID string `json:"key-id"` + WrappedKey []byte `json:"wrapped-key"` + NoncePrefix []byte `json:"nonce-prefix"` + BlockSize int `json:"block-size"` + PlaintextLength int64 `json:"plaintext-length"` +} + +// StandardEncryptionManager is a generic, format-agnostic [EncryptionManager] +// that provides envelope encryption for arbitrary files (e.g. manifests, +// manifest lists, Puffin statistics) using a [KeyManagementClient] to wrap +// and unwrap a fresh AES-256-GCM data encryption key (DEK) per file. +// +// Each file is split into fixed-size plaintext blocks, and each block is +// sealed independently with AES-GCM using a unique nonce (a per-file random +// prefix combined with the block index). This bounds memory usage and +// supports random access (Seek/ReadAt) on the decrypted file without +// buffering or decrypting more than the requested blocks. +// +// StandardEncryptionManager always encrypts and always decrypts: it fails +// closed, returning [ErrKeyIDRequired] or [ErrKeyMetadataRequired] rather +// than silently falling back to plaintext. Use [PlaintextEncryptionManager] +// for tables or files that are not encrypted. +type StandardEncryptionManager struct { + kms KeyManagementClient + dekLength int + blockSize int +} + +var _ EncryptionManager = (*StandardEncryptionManager)(nil) + +// StandardManagerOption configures a [StandardEncryptionManager] created by +// [NewStandardEncryptionManager]. +type StandardManagerOption func(*StandardEncryptionManager) + +// WithDEKLength overrides the default data encryption key length (in bytes). +// Valid AES key lengths are 16, 24, or 32 bytes. +func WithDEKLength(length int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.dekLength = length } +} + +// WithBlockSize overrides the default plaintext block size (in bytes) used +// to split files for independent block-level authentication. +func WithBlockSize(size int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.blockSize = size } +} + +// NewStandardEncryptionManager creates a [StandardEncryptionManager] backed +// by kms. kms must not be nil. +func NewStandardEncryptionManager(kms KeyManagementClient, opts ...StandardManagerOption) *StandardEncryptionManager { + m := &StandardEncryptionManager{ + kms: kms, + dekLength: StandardDefaultDEKLength, + blockSize: StandardDefaultBlockSize, + } + for _, opt := range opts { + opt(m) + } + + return m +} + +// NewEncryptedOutputFile creates a new AES-GCM block-encrypted output file. +// keyID identifies the KEK used to wrap the freshly generated per-file DEK, +// and must be non-empty; otherwise [ErrKeyIDRequired] is returned. +func (m *StandardEncryptionManager) NewEncryptedOutputFile(ctx context.Context, writer icebergio.FileWriter, keyID string) (EncryptedOutputFile, error) { + if keyID == "" { + return nil, ErrKeyIDRequired + } + if m.blockSize <= 0 { + return nil, fmt.Errorf("%w: got %d", ErrInvalidBlockSize, m.blockSize) + } + + var ( + plainDEK, wrappedDEK []byte + err error + ) + if m.kms.SupportsKeyGeneration() { + plainDEK, wrappedDEK, err = m.kms.GenerateKey(ctx, keyID, m.dekLength) + if err != nil { + return nil, fmt.Errorf("encryption: failed to generate DEK: %w", err) + } + } else { + plainDEK = make([]byte, m.dekLength) + if _, err = rand.Read(plainDEK); err != nil { + return nil, fmt.Errorf("encryption: failed to generate DEK: %w", err) + } + if wrappedDEK, err = m.kms.WrapKey(ctx, keyID, plainDEK); err != nil { + return nil, fmt.Errorf("encryption: failed to wrap DEK: %w", err) + } + } + + aead, err := newStandardAEAD(plainDEK) + if err != nil { + return nil, err + } + + noncePrefix := make([]byte, 4) + if _, err := rand.Read(noncePrefix); err != nil { + return nil, fmt.Errorf("encryption: failed to generate nonce prefix: %w", err) + } + + return &standardOutputFile{ + FileWriter: writer, + aead: aead, + noncePrefix: noncePrefix, + blockSize: m.blockSize, + keyID: keyID, + wrappedKey: wrappedDEK, + }, nil +} + +// NewDecryptedInputFile wraps file for transparent block-level AES-GCM +// decryption. keyMetadata must be the non-empty blob produced by +// [StandardEncryptionManager.NewEncryptedOutputFile]; otherwise +// [ErrKeyMetadataRequired] is returned. +func (m *StandardEncryptionManager) NewDecryptedInputFile(ctx context.Context, file icebergio.File, keyMetadata EncryptionKeyMetadata) (EncryptedInputFile, error) { + if len(keyMetadata) == 0 { + return nil, ErrKeyMetadataRequired + } + + var meta standardKeyMetadata + if err := json.Unmarshal(keyMetadata, &meta); err != nil { + return nil, fmt.Errorf("encryption: failed to decode key metadata: %w", err) + } + if meta.Version != standardKeyMetadataVersion { + return nil, fmt.Errorf("%w: %d", ErrUnsupportedKeyMetadataVersion, meta.Version) + } + if meta.BlockSize <= 0 { + return nil, fmt.Errorf("%w: block-size must be positive, got %d", ErrInvalidKeyMetadata, meta.BlockSize) + } + if meta.PlaintextLength < 0 { + return nil, fmt.Errorf("%w: plaintext-length must be non-negative, got %d", ErrInvalidKeyMetadata, meta.PlaintextLength) + } + if len(meta.NoncePrefix) != 4 { + return nil, fmt.Errorf("%w: nonce-prefix must be 4 bytes, got %d", ErrInvalidKeyMetadata, len(meta.NoncePrefix)) + } + + plainDEK, err := m.kms.UnwrapKey(ctx, meta.KeyID, meta.WrappedKey) + if err != nil { + return nil, fmt.Errorf("encryption: failed to unwrap DEK: %w", err) + } + + aead, err := newStandardAEAD(plainDEK) + if err != nil { + return nil, err + } + + return &standardInputFile{ + underlying: file, + aead: aead, + noncePrefix: meta.NoncePrefix, + blockSize: meta.BlockSize, + plaintextLength: meta.PlaintextLength, + keyMetadata: keyMetadata, + }, nil +} + +func newStandardAEAD(key []byte) (cipher.AEAD, error) { + block, err := aes.NewCipher(key) + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrInvalidKeyLength, err) + } + gcm, err := cipher.NewGCM(block) + if err != nil { + return nil, fmt.Errorf("encryption: failed to create GCM: %w", err) + } + + return gcm, nil +} + +// standardBlockNonce derives the AES-GCM nonce for blockIndex: the 4-byte +// per-file random prefix followed by the 8-byte big-endian block index. The +// (key, nonce) pair is unique per block since the DEK is fresh per file and +// no two blocks in the same file share an index. +func standardBlockNonce(prefix []byte, blockIndex uint64) []byte { + nonce := make([]byte, 12) + copy(nonce, prefix) + binary.BigEndian.PutUint64(nonce[4:], blockIndex) + + return nonce +} + +// standardOutputFile is an [EncryptedOutputFile] that seals fixed-size +// plaintext blocks with AES-GCM as they are written. +type standardOutputFile struct { + icebergio.FileWriter + + aead cipher.AEAD + noncePrefix []byte + blockSize int + keyID string + wrappedKey []byte + + buf []byte + blockIndex uint64 + written int64 + closed bool + + keyMetadata EncryptionKeyMetadata +} + +var _ EncryptedOutputFile = (*standardOutputFile)(nil) + +func (f *standardOutputFile) Write(p []byte) (int, error) { + total := len(p) + for len(p) > 0 { + space := f.blockSize - len(f.buf) + n := min(space, len(p)) + f.buf = append(f.buf, p[:n]...) + p = p[n:] + f.written += int64(n) + if len(f.buf) == f.blockSize { + if err := f.flushBlock(); err != nil { + return total - len(p), err + } + } + } + + return total, nil +} + +func (f *standardOutputFile) flushBlock() error { + ciphertext := f.aead.Seal(nil, standardBlockNonce(f.noncePrefix, f.blockIndex), f.buf, nil) + if _, err := f.FileWriter.Write(ciphertext); err != nil { + return fmt.Errorf("encryption: failed to write encrypted block: %w", err) + } + f.blockIndex++ + f.buf = f.buf[:0] + + return nil +} + +// ReadFrom copies from r, encrypting as data is written, satisfying +// io.ReaderFrom (required by [icebergio.FileWriter]). +func (f *standardOutputFile) ReadFrom(r io.Reader) (int64, error) { + buf := make([]byte, 32*1024) + var total int64 + for { + n, err := r.Read(buf) + if n > 0 { + wn, werr := f.Write(buf[:n]) + total += int64(wn) + if werr != nil { + return total, werr + } + } + if err == io.EOF { + break + } + if err != nil { + return total, err + } + } + + return total, nil +} + +func (f *standardOutputFile) Close() error { + if f.closed { + return nil + } + f.closed = true + + if len(f.buf) > 0 { + if err := f.flushBlock(); err != nil { + _ = f.FileWriter.Close() + + return err + } + } + + meta := standardKeyMetadata{ + Version: standardKeyMetadataVersion, + KeyID: f.keyID, + WrappedKey: f.wrappedKey, + NoncePrefix: f.noncePrefix, + BlockSize: f.blockSize, + PlaintextLength: f.written, + } + encoded, err := json.Marshal(meta) + if err != nil { + _ = f.FileWriter.Close() + + return fmt.Errorf("encryption: failed to encode key metadata: %w", err) + } + f.keyMetadata = encoded + + return f.FileWriter.Close() +} + +// KeyMetadata returns the finalized per-file key metadata. It is only +// populated after Close has been called. +func (f *standardOutputFile) KeyMetadata() EncryptionKeyMetadata { return f.keyMetadata } + +// standardInputFile is an [EncryptedInputFile] that decrypts fixed-size +// AES-GCM blocks on demand, supporting random access via ReadAt/Seek. +type standardInputFile struct { + underlying icebergio.File + aead cipher.AEAD + noncePrefix []byte + blockSize int + plaintextLength int64 + keyMetadata EncryptionKeyMetadata + + pos int64 +} + +var _ EncryptedInputFile = (*standardInputFile)(nil) + +func (f *standardInputFile) numBlocks() int64 { + if f.plaintextLength == 0 { + return 0 + } + + return (f.plaintextLength + int64(f.blockSize) - 1) / int64(f.blockSize) +} + +func (f *standardInputFile) blockPlainLen(idx int64) int64 { + if idx == f.numBlocks()-1 { + return f.plaintextLength - idx*int64(f.blockSize) + } + + return int64(f.blockSize) +} + +func (f *standardInputFile) physicalOffset(idx int64) int64 { + return idx * int64(f.blockSize+f.aead.Overhead()) +} + +func (f *standardInputFile) readBlock(idx int64) ([]byte, error) { + plainLen := f.blockPlainLen(idx) + ciphertext := make([]byte, plainLen+int64(f.aead.Overhead())) + if _, err := f.underlying.ReadAt(ciphertext, f.physicalOffset(idx)); err != nil && !errors.Is(err, io.EOF) { + return nil, fmt.Errorf("encryption: failed to read block %d: %w", idx, err) + } + + plaintext, err := f.aead.Open(nil, standardBlockNonce(f.noncePrefix, uint64(idx)), ciphertext, nil) + if err != nil { + return nil, fmt.Errorf("%w: block %d: %v", ErrAuthenticationFailed, idx, err) + } + + return plaintext, nil +} + +func (f *standardInputFile) ReadAt(p []byte, off int64) (int, error) { + if off < 0 { + return 0, errors.New("encryption: ReadAt: negative offset") + } + if off >= f.plaintextLength { + return 0, io.EOF + } + + var read int Review Comment: doc nit: `ReadAt` is stateless and safe for concurrent use (and `io.ReaderAt` documents that contract), but `Read`/`Seek` mutate `f.pos` with no synchronization. Same type mixing both is a bit of a trap via `io.Copy`. Worth a comment that `Read`/`Seek` aren't concurrent-safe while `ReadAt` is. ########## encryption/standard_manager.go: ########## @@ -0,0 +1,511 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package encryption + +import ( + "context" + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "io/fs" + + icebergio "github.com/apache/iceberg-go/io" +) + +// Defaults for [StandardEncryptionManager]. +const ( + // StandardDefaultDEKLength is the default length, in bytes, of the + // per-file data encryption key (DEK) generated for AES-256-GCM. + StandardDefaultDEKLength = 32 + + // StandardDefaultBlockSize is the default plaintext block size, in + // bytes, used to split a file into independently authenticated AES-GCM + // blocks. Blocks allow random access (Seek/ReadAt) without buffering or + // decrypting the whole file. + StandardDefaultBlockSize = 64 * 1024 +) + +// Sentinel errors returned by [StandardEncryptionManager]. +var ( + // ErrKeyIDRequired is returned by + // [StandardEncryptionManager.NewEncryptedOutputFile] when keyID is empty. + // StandardEncryptionManager always encrypts, so it requires a KEK to wrap + // the generated DEK; use [PlaintextEncryptionManager] for unencrypted + // tables instead of passing an empty keyID here. + ErrKeyIDRequired = errors.New("encryption: StandardEncryptionManager requires a non-empty keyID") + + // ErrKeyMetadataRequired is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when keyMetadata is + // empty. StandardEncryptionManager always decrypts, so it requires the + // per-file key metadata produced by [StandardEncryptionManager.NewEncryptedOutputFile]. + ErrKeyMetadataRequired = errors.New("encryption: StandardEncryptionManager requires non-empty key metadata") + + // ErrUnsupportedKeyMetadataVersion is returned when key metadata was + // produced by a newer, incompatible encoding version. + ErrUnsupportedKeyMetadataVersion = errors.New("encryption: unsupported key metadata version") + + // ErrInvalidBlockSize is returned when a configured or decoded block + // size is not positive. + ErrInvalidBlockSize = errors.New("encryption: block size must be positive") + + // ErrInvalidKeyMetadata is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when decoded key + // metadata fails basic sanity checks (e.g. a negative plaintext length + // or a nonce prefix of the wrong size). Key metadata is untrusted input + // on a crypto read path, so it is validated rather than trusted blindly. + ErrInvalidKeyMetadata = errors.New("encryption: invalid key metadata") +) + +// standardKeyMetadataVersion is the current encoding version written by +// [StandardEncryptionManager]. It is bumped whenever the on-disk layout of +// standardKeyMetadata or the block ciphertext format changes incompatibly. +const standardKeyMetadataVersion = 1 + +// standardKeyMetadata is the JSON-encoded structure stored as the opaque +// [EncryptionKeyMetadata] for files produced by [StandardEncryptionManager]. +type standardKeyMetadata struct { + Version int `json:"v"` + KeyID string `json:"key-id"` + WrappedKey []byte `json:"wrapped-key"` + NoncePrefix []byte `json:"nonce-prefix"` + BlockSize int `json:"block-size"` + PlaintextLength int64 `json:"plaintext-length"` +} + +// StandardEncryptionManager is a generic, format-agnostic [EncryptionManager] +// that provides envelope encryption for arbitrary files (e.g. manifests, +// manifest lists, Puffin statistics) using a [KeyManagementClient] to wrap +// and unwrap a fresh AES-256-GCM data encryption key (DEK) per file. +// +// Each file is split into fixed-size plaintext blocks, and each block is +// sealed independently with AES-GCM using a unique nonce (a per-file random +// prefix combined with the block index). This bounds memory usage and +// supports random access (Seek/ReadAt) on the decrypted file without +// buffering or decrypting more than the requested blocks. +// +// StandardEncryptionManager always encrypts and always decrypts: it fails +// closed, returning [ErrKeyIDRequired] or [ErrKeyMetadataRequired] rather +// than silently falling back to plaintext. Use [PlaintextEncryptionManager] +// for tables or files that are not encrypted. +type StandardEncryptionManager struct { + kms KeyManagementClient + dekLength int + blockSize int +} + +var _ EncryptionManager = (*StandardEncryptionManager)(nil) + +// StandardManagerOption configures a [StandardEncryptionManager] created by +// [NewStandardEncryptionManager]. +type StandardManagerOption func(*StandardEncryptionManager) + +// WithDEKLength overrides the default data encryption key length (in bytes). +// Valid AES key lengths are 16, 24, or 32 bytes. +func WithDEKLength(length int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.dekLength = length } +} + +// WithBlockSize overrides the default plaintext block size (in bytes) used +// to split files for independent block-level authentication. +func WithBlockSize(size int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.blockSize = size } +} + +// NewStandardEncryptionManager creates a [StandardEncryptionManager] backed +// by kms. kms must not be nil. +func NewStandardEncryptionManager(kms KeyManagementClient, opts ...StandardManagerOption) *StandardEncryptionManager { + m := &StandardEncryptionManager{ + kms: kms, + dekLength: StandardDefaultDEKLength, + blockSize: StandardDefaultBlockSize, + } + for _, opt := range opts { + opt(m) + } + + return m +} + +// NewEncryptedOutputFile creates a new AES-GCM block-encrypted output file. +// keyID identifies the KEK used to wrap the freshly generated per-file DEK, +// and must be non-empty; otherwise [ErrKeyIDRequired] is returned. +func (m *StandardEncryptionManager) NewEncryptedOutputFile(ctx context.Context, writer icebergio.FileWriter, keyID string) (EncryptedOutputFile, error) { + if keyID == "" { + return nil, ErrKeyIDRequired + } + if m.blockSize <= 0 { + return nil, fmt.Errorf("%w: got %d", ErrInvalidBlockSize, m.blockSize) + } + + var ( + plainDEK, wrappedDEK []byte + err error + ) + if m.kms.SupportsKeyGeneration() { + plainDEK, wrappedDEK, err = m.kms.GenerateKey(ctx, keyID, m.dekLength) + if err != nil { + return nil, fmt.Errorf("encryption: failed to generate DEK: %w", err) + } + } else { + plainDEK = make([]byte, m.dekLength) + if _, err = rand.Read(plainDEK); err != nil { + return nil, fmt.Errorf("encryption: failed to generate DEK: %w", err) + } + if wrappedDEK, err = m.kms.WrapKey(ctx, keyID, plainDEK); err != nil { + return nil, fmt.Errorf("encryption: failed to wrap DEK: %w", err) + } + } + + aead, err := newStandardAEAD(plainDEK) + if err != nil { + return nil, err + } + + noncePrefix := make([]byte, 4) + if _, err := rand.Read(noncePrefix); err != nil { + return nil, fmt.Errorf("encryption: failed to generate nonce prefix: %w", err) + } + + return &standardOutputFile{ + FileWriter: writer, + aead: aead, + noncePrefix: noncePrefix, + blockSize: m.blockSize, + keyID: keyID, + wrappedKey: wrappedDEK, + }, nil +} + +// NewDecryptedInputFile wraps file for transparent block-level AES-GCM +// decryption. keyMetadata must be the non-empty blob produced by +// [StandardEncryptionManager.NewEncryptedOutputFile]; otherwise +// [ErrKeyMetadataRequired] is returned. +func (m *StandardEncryptionManager) NewDecryptedInputFile(ctx context.Context, file icebergio.File, keyMetadata EncryptionKeyMetadata) (EncryptedInputFile, error) { + if len(keyMetadata) == 0 { + return nil, ErrKeyMetadataRequired + } + + var meta standardKeyMetadata + if err := json.Unmarshal(keyMetadata, &meta); err != nil { + return nil, fmt.Errorf("encryption: failed to decode key metadata: %w", err) + } + if meta.Version != standardKeyMetadataVersion { + return nil, fmt.Errorf("%w: %d", ErrUnsupportedKeyMetadataVersion, meta.Version) + } + if meta.BlockSize <= 0 { + return nil, fmt.Errorf("%w: block-size must be positive, got %d", ErrInvalidKeyMetadata, meta.BlockSize) + } + if meta.PlaintextLength < 0 { + return nil, fmt.Errorf("%w: plaintext-length must be non-negative, got %d", ErrInvalidKeyMetadata, meta.PlaintextLength) + } + if len(meta.NoncePrefix) != 4 { + return nil, fmt.Errorf("%w: nonce-prefix must be 4 bytes, got %d", ErrInvalidKeyMetadata, len(meta.NoncePrefix)) + } + + plainDEK, err := m.kms.UnwrapKey(ctx, meta.KeyID, meta.WrappedKey) Review Comment: tiny consistency nit: this wraps with `%v` while the `NewGCM` error just below uses `%w`. `errors.Is(ErrInvalidKeyLength)` still works, but `%v` drops the inner aes error from the chain. I'd use `%w` for both. ########## encryption/standard_manager.go: ########## @@ -0,0 +1,511 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package encryption + +import ( + "context" + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "io/fs" + + icebergio "github.com/apache/iceberg-go/io" +) + +// Defaults for [StandardEncryptionManager]. +const ( + // StandardDefaultDEKLength is the default length, in bytes, of the + // per-file data encryption key (DEK) generated for AES-256-GCM. + StandardDefaultDEKLength = 32 + + // StandardDefaultBlockSize is the default plaintext block size, in + // bytes, used to split a file into independently authenticated AES-GCM + // blocks. Blocks allow random access (Seek/ReadAt) without buffering or + // decrypting the whole file. + StandardDefaultBlockSize = 64 * 1024 +) + +// Sentinel errors returned by [StandardEncryptionManager]. +var ( + // ErrKeyIDRequired is returned by + // [StandardEncryptionManager.NewEncryptedOutputFile] when keyID is empty. + // StandardEncryptionManager always encrypts, so it requires a KEK to wrap + // the generated DEK; use [PlaintextEncryptionManager] for unencrypted + // tables instead of passing an empty keyID here. + ErrKeyIDRequired = errors.New("encryption: StandardEncryptionManager requires a non-empty keyID") + + // ErrKeyMetadataRequired is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when keyMetadata is + // empty. StandardEncryptionManager always decrypts, so it requires the + // per-file key metadata produced by [StandardEncryptionManager.NewEncryptedOutputFile]. + ErrKeyMetadataRequired = errors.New("encryption: StandardEncryptionManager requires non-empty key metadata") + + // ErrUnsupportedKeyMetadataVersion is returned when key metadata was + // produced by a newer, incompatible encoding version. + ErrUnsupportedKeyMetadataVersion = errors.New("encryption: unsupported key metadata version") + + // ErrInvalidBlockSize is returned when a configured or decoded block + // size is not positive. + ErrInvalidBlockSize = errors.New("encryption: block size must be positive") + + // ErrInvalidKeyMetadata is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when decoded key + // metadata fails basic sanity checks (e.g. a negative plaintext length + // or a nonce prefix of the wrong size). Key metadata is untrusted input + // on a crypto read path, so it is validated rather than trusted blindly. + ErrInvalidKeyMetadata = errors.New("encryption: invalid key metadata") +) + +// standardKeyMetadataVersion is the current encoding version written by +// [StandardEncryptionManager]. It is bumped whenever the on-disk layout of +// standardKeyMetadata or the block ciphertext format changes incompatibly. +const standardKeyMetadataVersion = 1 + +// standardKeyMetadata is the JSON-encoded structure stored as the opaque +// [EncryptionKeyMetadata] for files produced by [StandardEncryptionManager]. +type standardKeyMetadata struct { + Version int `json:"v"` + KeyID string `json:"key-id"` + WrappedKey []byte `json:"wrapped-key"` + NoncePrefix []byte `json:"nonce-prefix"` + BlockSize int `json:"block-size"` + PlaintextLength int64 `json:"plaintext-length"` +} + +// StandardEncryptionManager is a generic, format-agnostic [EncryptionManager] +// that provides envelope encryption for arbitrary files (e.g. manifests, +// manifest lists, Puffin statistics) using a [KeyManagementClient] to wrap +// and unwrap a fresh AES-256-GCM data encryption key (DEK) per file. +// +// Each file is split into fixed-size plaintext blocks, and each block is +// sealed independently with AES-GCM using a unique nonce (a per-file random +// prefix combined with the block index). This bounds memory usage and +// supports random access (Seek/ReadAt) on the decrypted file without +// buffering or decrypting more than the requested blocks. +// +// StandardEncryptionManager always encrypts and always decrypts: it fails +// closed, returning [ErrKeyIDRequired] or [ErrKeyMetadataRequired] rather +// than silently falling back to plaintext. Use [PlaintextEncryptionManager] +// for tables or files that are not encrypted. +type StandardEncryptionManager struct { + kms KeyManagementClient + dekLength int + blockSize int +} + +var _ EncryptionManager = (*StandardEncryptionManager)(nil) + +// StandardManagerOption configures a [StandardEncryptionManager] created by +// [NewStandardEncryptionManager]. +type StandardManagerOption func(*StandardEncryptionManager) + +// WithDEKLength overrides the default data encryption key length (in bytes). +// Valid AES key lengths are 16, 24, or 32 bytes. +func WithDEKLength(length int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.dekLength = length } +} + +// WithBlockSize overrides the default plaintext block size (in bytes) used +// to split files for independent block-level authentication. +func WithBlockSize(size int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.blockSize = size } +} + +// NewStandardEncryptionManager creates a [StandardEncryptionManager] backed +// by kms. kms must not be nil. +func NewStandardEncryptionManager(kms KeyManagementClient, opts ...StandardManagerOption) *StandardEncryptionManager { + m := &StandardEncryptionManager{ + kms: kms, + dekLength: StandardDefaultDEKLength, + blockSize: StandardDefaultBlockSize, + } + for _, opt := range opts { + opt(m) + } + + return m +} + +// NewEncryptedOutputFile creates a new AES-GCM block-encrypted output file. +// keyID identifies the KEK used to wrap the freshly generated per-file DEK, +// and must be non-empty; otherwise [ErrKeyIDRequired] is returned. +func (m *StandardEncryptionManager) NewEncryptedOutputFile(ctx context.Context, writer icebergio.FileWriter, keyID string) (EncryptedOutputFile, error) { + if keyID == "" { + return nil, ErrKeyIDRequired + } + if m.blockSize <= 0 { + return nil, fmt.Errorf("%w: got %d", ErrInvalidBlockSize, m.blockSize) + } + + var ( + plainDEK, wrappedDEK []byte + err error + ) + if m.kms.SupportsKeyGeneration() { + plainDEK, wrappedDEK, err = m.kms.GenerateKey(ctx, keyID, m.dekLength) + if err != nil { + return nil, fmt.Errorf("encryption: failed to generate DEK: %w", err) + } + } else { + plainDEK = make([]byte, m.dekLength) + if _, err = rand.Read(plainDEK); err != nil { + return nil, fmt.Errorf("encryption: failed to generate DEK: %w", err) + } + if wrappedDEK, err = m.kms.WrapKey(ctx, keyID, plainDEK); err != nil { + return nil, fmt.Errorf("encryption: failed to wrap DEK: %w", err) + } + } + + aead, err := newStandardAEAD(plainDEK) + if err != nil { + return nil, err + } + + noncePrefix := make([]byte, 4) + if _, err := rand.Read(noncePrefix); err != nil { + return nil, fmt.Errorf("encryption: failed to generate nonce prefix: %w", err) + } + + return &standardOutputFile{ + FileWriter: writer, + aead: aead, + noncePrefix: noncePrefix, + blockSize: m.blockSize, + keyID: keyID, + wrappedKey: wrappedDEK, + }, nil +} + +// NewDecryptedInputFile wraps file for transparent block-level AES-GCM +// decryption. keyMetadata must be the non-empty blob produced by +// [StandardEncryptionManager.NewEncryptedOutputFile]; otherwise +// [ErrKeyMetadataRequired] is returned. +func (m *StandardEncryptionManager) NewDecryptedInputFile(ctx context.Context, file icebergio.File, keyMetadata EncryptionKeyMetadata) (EncryptedInputFile, error) { + if len(keyMetadata) == 0 { + return nil, ErrKeyMetadataRequired + } + + var meta standardKeyMetadata + if err := json.Unmarshal(keyMetadata, &meta); err != nil { + return nil, fmt.Errorf("encryption: failed to decode key metadata: %w", err) + } + if meta.Version != standardKeyMetadataVersion { + return nil, fmt.Errorf("%w: %d", ErrUnsupportedKeyMetadataVersion, meta.Version) + } + if meta.BlockSize <= 0 { + return nil, fmt.Errorf("%w: block-size must be positive, got %d", ErrInvalidKeyMetadata, meta.BlockSize) + } + if meta.PlaintextLength < 0 { + return nil, fmt.Errorf("%w: plaintext-length must be non-negative, got %d", ErrInvalidKeyMetadata, meta.PlaintextLength) + } + if len(meta.NoncePrefix) != 4 { + return nil, fmt.Errorf("%w: nonce-prefix must be 4 bytes, got %d", ErrInvalidKeyMetadata, len(meta.NoncePrefix)) + } + + plainDEK, err := m.kms.UnwrapKey(ctx, meta.KeyID, meta.WrappedKey) + if err != nil { + return nil, fmt.Errorf("encryption: failed to unwrap DEK: %w", err) + } + + aead, err := newStandardAEAD(plainDEK) + if err != nil { + return nil, err + } + + return &standardInputFile{ + underlying: file, + aead: aead, + noncePrefix: meta.NoncePrefix, + blockSize: meta.BlockSize, + plaintextLength: meta.PlaintextLength, + keyMetadata: keyMetadata, + }, nil +} + +func newStandardAEAD(key []byte) (cipher.AEAD, error) { + block, err := aes.NewCipher(key) + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrInvalidKeyLength, err) + } + gcm, err := cipher.NewGCM(block) + if err != nil { + return nil, fmt.Errorf("encryption: failed to create GCM: %w", err) + } + + return gcm, nil +} + +// standardBlockNonce derives the AES-GCM nonce for blockIndex: the 4-byte +// per-file random prefix followed by the 8-byte big-endian block index. The +// (key, nonce) pair is unique per block since the DEK is fresh per file and +// no two blocks in the same file share an index. +func standardBlockNonce(prefix []byte, blockIndex uint64) []byte { + nonce := make([]byte, 12) + copy(nonce, prefix) + binary.BigEndian.PutUint64(nonce[4:], blockIndex) + + return nonce +} + +// standardOutputFile is an [EncryptedOutputFile] that seals fixed-size +// plaintext blocks with AES-GCM as they are written. +type standardOutputFile struct { + icebergio.FileWriter + + aead cipher.AEAD + noncePrefix []byte + blockSize int + keyID string + wrappedKey []byte + + buf []byte + blockIndex uint64 + written int64 + closed bool + + keyMetadata EncryptionKeyMetadata +} + +var _ EncryptedOutputFile = (*standardOutputFile)(nil) + +func (f *standardOutputFile) Write(p []byte) (int, error) { + total := len(p) + for len(p) > 0 { + space := f.blockSize - len(f.buf) + n := min(space, len(p)) + f.buf = append(f.buf, p[:n]...) + p = p[n:] + f.written += int64(n) + if len(f.buf) == f.blockSize { + if err := f.flushBlock(); err != nil { Review Comment: nit: the copy buffer is a fixed 32 KiB regardless of `blockSize`, so with a larger block we do extra Write round-trips through the buffering. `make([]byte, f.blockSize)` (or `max(32*1024, f.blockSize)`) lines it up with the block boundary. ########## encryption/standard_manager.go: ########## @@ -0,0 +1,511 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package encryption + +import ( + "context" + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "io/fs" + + icebergio "github.com/apache/iceberg-go/io" +) + +// Defaults for [StandardEncryptionManager]. +const ( + // StandardDefaultDEKLength is the default length, in bytes, of the + // per-file data encryption key (DEK) generated for AES-256-GCM. + StandardDefaultDEKLength = 32 + + // StandardDefaultBlockSize is the default plaintext block size, in + // bytes, used to split a file into independently authenticated AES-GCM + // blocks. Blocks allow random access (Seek/ReadAt) without buffering or + // decrypting the whole file. + StandardDefaultBlockSize = 64 * 1024 +) + +// Sentinel errors returned by [StandardEncryptionManager]. +var ( + // ErrKeyIDRequired is returned by + // [StandardEncryptionManager.NewEncryptedOutputFile] when keyID is empty. + // StandardEncryptionManager always encrypts, so it requires a KEK to wrap + // the generated DEK; use [PlaintextEncryptionManager] for unencrypted + // tables instead of passing an empty keyID here. + ErrKeyIDRequired = errors.New("encryption: StandardEncryptionManager requires a non-empty keyID") + + // ErrKeyMetadataRequired is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when keyMetadata is + // empty. StandardEncryptionManager always decrypts, so it requires the + // per-file key metadata produced by [StandardEncryptionManager.NewEncryptedOutputFile]. + ErrKeyMetadataRequired = errors.New("encryption: StandardEncryptionManager requires non-empty key metadata") + + // ErrUnsupportedKeyMetadataVersion is returned when key metadata was + // produced by a newer, incompatible encoding version. + ErrUnsupportedKeyMetadataVersion = errors.New("encryption: unsupported key metadata version") + + // ErrInvalidBlockSize is returned when a configured or decoded block + // size is not positive. + ErrInvalidBlockSize = errors.New("encryption: block size must be positive") + + // ErrInvalidKeyMetadata is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when decoded key + // metadata fails basic sanity checks (e.g. a negative plaintext length + // or a nonce prefix of the wrong size). Key metadata is untrusted input + // on a crypto read path, so it is validated rather than trusted blindly. + ErrInvalidKeyMetadata = errors.New("encryption: invalid key metadata") +) + +// standardKeyMetadataVersion is the current encoding version written by +// [StandardEncryptionManager]. It is bumped whenever the on-disk layout of +// standardKeyMetadata or the block ciphertext format changes incompatibly. +const standardKeyMetadataVersion = 1 + +// standardKeyMetadata is the JSON-encoded structure stored as the opaque +// [EncryptionKeyMetadata] for files produced by [StandardEncryptionManager]. +type standardKeyMetadata struct { + Version int `json:"v"` + KeyID string `json:"key-id"` + WrappedKey []byte `json:"wrapped-key"` + NoncePrefix []byte `json:"nonce-prefix"` + BlockSize int `json:"block-size"` + PlaintextLength int64 `json:"plaintext-length"` +} + +// StandardEncryptionManager is a generic, format-agnostic [EncryptionManager] +// that provides envelope encryption for arbitrary files (e.g. manifests, +// manifest lists, Puffin statistics) using a [KeyManagementClient] to wrap +// and unwrap a fresh AES-256-GCM data encryption key (DEK) per file. +// +// Each file is split into fixed-size plaintext blocks, and each block is +// sealed independently with AES-GCM using a unique nonce (a per-file random +// prefix combined with the block index). This bounds memory usage and +// supports random access (Seek/ReadAt) on the decrypted file without +// buffering or decrypting more than the requested blocks. +// +// StandardEncryptionManager always encrypts and always decrypts: it fails +// closed, returning [ErrKeyIDRequired] or [ErrKeyMetadataRequired] rather +// than silently falling back to plaintext. Use [PlaintextEncryptionManager] +// for tables or files that are not encrypted. +type StandardEncryptionManager struct { + kms KeyManagementClient + dekLength int + blockSize int +} + +var _ EncryptionManager = (*StandardEncryptionManager)(nil) + +// StandardManagerOption configures a [StandardEncryptionManager] created by +// [NewStandardEncryptionManager]. +type StandardManagerOption func(*StandardEncryptionManager) + +// WithDEKLength overrides the default data encryption key length (in bytes). +// Valid AES key lengths are 16, 24, or 32 bytes. +func WithDEKLength(length int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.dekLength = length } +} + +// WithBlockSize overrides the default plaintext block size (in bytes) used +// to split files for independent block-level authentication. +func WithBlockSize(size int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.blockSize = size } +} + +// NewStandardEncryptionManager creates a [StandardEncryptionManager] backed +// by kms. kms must not be nil. +func NewStandardEncryptionManager(kms KeyManagementClient, opts ...StandardManagerOption) *StandardEncryptionManager { + m := &StandardEncryptionManager{ + kms: kms, + dekLength: StandardDefaultDEKLength, + blockSize: StandardDefaultBlockSize, + } + for _, opt := range opts { + opt(m) + } + + return m +} + +// NewEncryptedOutputFile creates a new AES-GCM block-encrypted output file. +// keyID identifies the KEK used to wrap the freshly generated per-file DEK, +// and must be non-empty; otherwise [ErrKeyIDRequired] is returned. +func (m *StandardEncryptionManager) NewEncryptedOutputFile(ctx context.Context, writer icebergio.FileWriter, keyID string) (EncryptedOutputFile, error) { + if keyID == "" { + return nil, ErrKeyIDRequired + } + if m.blockSize <= 0 { + return nil, fmt.Errorf("%w: got %d", ErrInvalidBlockSize, m.blockSize) + } + + var ( + plainDEK, wrappedDEK []byte + err error + ) + if m.kms.SupportsKeyGeneration() { + plainDEK, wrappedDEK, err = m.kms.GenerateKey(ctx, keyID, m.dekLength) + if err != nil { + return nil, fmt.Errorf("encryption: failed to generate DEK: %w", err) + } + } else { + plainDEK = make([]byte, m.dekLength) + if _, err = rand.Read(plainDEK); err != nil { + return nil, fmt.Errorf("encryption: failed to generate DEK: %w", err) + } + if wrappedDEK, err = m.kms.WrapKey(ctx, keyID, plainDEK); err != nil { + return nil, fmt.Errorf("encryption: failed to wrap DEK: %w", err) + } + } + + aead, err := newStandardAEAD(plainDEK) + if err != nil { + return nil, err + } + + noncePrefix := make([]byte, 4) + if _, err := rand.Read(noncePrefix); err != nil { + return nil, fmt.Errorf("encryption: failed to generate nonce prefix: %w", err) + } + + return &standardOutputFile{ + FileWriter: writer, + aead: aead, + noncePrefix: noncePrefix, + blockSize: m.blockSize, + keyID: keyID, + wrappedKey: wrappedDEK, + }, nil +} + +// NewDecryptedInputFile wraps file for transparent block-level AES-GCM +// decryption. keyMetadata must be the non-empty blob produced by +// [StandardEncryptionManager.NewEncryptedOutputFile]; otherwise +// [ErrKeyMetadataRequired] is returned. +func (m *StandardEncryptionManager) NewDecryptedInputFile(ctx context.Context, file icebergio.File, keyMetadata EncryptionKeyMetadata) (EncryptedInputFile, error) { + if len(keyMetadata) == 0 { + return nil, ErrKeyMetadataRequired + } + + var meta standardKeyMetadata + if err := json.Unmarshal(keyMetadata, &meta); err != nil { + return nil, fmt.Errorf("encryption: failed to decode key metadata: %w", err) + } + if meta.Version != standardKeyMetadataVersion { + return nil, fmt.Errorf("%w: %d", ErrUnsupportedKeyMetadataVersion, meta.Version) + } + if meta.BlockSize <= 0 { + return nil, fmt.Errorf("%w: block-size must be positive, got %d", ErrInvalidKeyMetadata, meta.BlockSize) + } + if meta.PlaintextLength < 0 { + return nil, fmt.Errorf("%w: plaintext-length must be non-negative, got %d", ErrInvalidKeyMetadata, meta.PlaintextLength) + } + if len(meta.NoncePrefix) != 4 { + return nil, fmt.Errorf("%w: nonce-prefix must be 4 bytes, got %d", ErrInvalidKeyMetadata, len(meta.NoncePrefix)) + } + + plainDEK, err := m.kms.UnwrapKey(ctx, meta.KeyID, meta.WrappedKey) + if err != nil { + return nil, fmt.Errorf("encryption: failed to unwrap DEK: %w", err) + } + + aead, err := newStandardAEAD(plainDEK) + if err != nil { + return nil, err + } + + return &standardInputFile{ + underlying: file, + aead: aead, + noncePrefix: meta.NoncePrefix, + blockSize: meta.BlockSize, + plaintextLength: meta.PlaintextLength, + keyMetadata: keyMetadata, + }, nil +} + +func newStandardAEAD(key []byte) (cipher.AEAD, error) { + block, err := aes.NewCipher(key) + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrInvalidKeyLength, err) + } + gcm, err := cipher.NewGCM(block) + if err != nil { + return nil, fmt.Errorf("encryption: failed to create GCM: %w", err) + } + + return gcm, nil +} + +// standardBlockNonce derives the AES-GCM nonce for blockIndex: the 4-byte +// per-file random prefix followed by the 8-byte big-endian block index. The +// (key, nonce) pair is unique per block since the DEK is fresh per file and +// no two blocks in the same file share an index. +func standardBlockNonce(prefix []byte, blockIndex uint64) []byte { + nonce := make([]byte, 12) + copy(nonce, prefix) + binary.BigEndian.PutUint64(nonce[4:], blockIndex) + + return nonce +} + +// standardOutputFile is an [EncryptedOutputFile] that seals fixed-size +// plaintext blocks with AES-GCM as they are written. +type standardOutputFile struct { + icebergio.FileWriter + + aead cipher.AEAD + noncePrefix []byte + blockSize int + keyID string + wrappedKey []byte + + buf []byte + blockIndex uint64 + written int64 + closed bool + + keyMetadata EncryptionKeyMetadata +} + +var _ EncryptedOutputFile = (*standardOutputFile)(nil) + +func (f *standardOutputFile) Write(p []byte) (int, error) { + total := len(p) + for len(p) > 0 { + space := f.blockSize - len(f.buf) + n := min(space, len(p)) + f.buf = append(f.buf, p[:n]...) + p = p[n:] + f.written += int64(n) + if len(f.buf) == f.blockSize { + if err := f.flushBlock(); err != nil { + return total - len(p), err + } + } + } + + return total, nil +} + +func (f *standardOutputFile) flushBlock() error { + ciphertext := f.aead.Seal(nil, standardBlockNonce(f.noncePrefix, f.blockIndex), f.buf, nil) + if _, err := f.FileWriter.Write(ciphertext); err != nil { + return fmt.Errorf("encryption: failed to write encrypted block: %w", err) + } + f.blockIndex++ + f.buf = f.buf[:0] + + return nil +} + +// ReadFrom copies from r, encrypting as data is written, satisfying +// io.ReaderFrom (required by [icebergio.FileWriter]). +func (f *standardOutputFile) ReadFrom(r io.Reader) (int64, error) { + buf := make([]byte, 32*1024) + var total int64 + for { + n, err := r.Read(buf) + if n > 0 { + wn, werr := f.Write(buf[:n]) + total += int64(wn) + if werr != nil { + return total, werr + } + } + if err == io.EOF { + break + } + if err != nil { + return total, err + } + } + + return total, nil +} + +func (f *standardOutputFile) Close() error { + if f.closed { + return nil + } + f.closed = true + + if len(f.buf) > 0 { + if err := f.flushBlock(); err != nil { + _ = f.FileWriter.Close() + + return err + } + } + + meta := standardKeyMetadata{ + Version: standardKeyMetadataVersion, + KeyID: f.keyID, + WrappedKey: f.wrappedKey, + NoncePrefix: f.noncePrefix, + BlockSize: f.blockSize, + PlaintextLength: f.written, + } + encoded, err := json.Marshal(meta) + if err != nil { + _ = f.FileWriter.Close() + + return fmt.Errorf("encryption: failed to encode key metadata: %w", err) + } + f.keyMetadata = encoded + + return f.FileWriter.Close() +} + +// KeyMetadata returns the finalized per-file key metadata. It is only +// populated after Close has been called. +func (f *standardOutputFile) KeyMetadata() EncryptionKeyMetadata { return f.keyMetadata } + +// standardInputFile is an [EncryptedInputFile] that decrypts fixed-size +// AES-GCM blocks on demand, supporting random access via ReadAt/Seek. +type standardInputFile struct { + underlying icebergio.File + aead cipher.AEAD + noncePrefix []byte + blockSize int + plaintextLength int64 + keyMetadata EncryptionKeyMetadata + + pos int64 +} + +var _ EncryptedInputFile = (*standardInputFile)(nil) + +func (f *standardInputFile) numBlocks() int64 { + if f.plaintextLength == 0 { + return 0 + } + + return (f.plaintextLength + int64(f.blockSize) - 1) / int64(f.blockSize) +} + +func (f *standardInputFile) blockPlainLen(idx int64) int64 { + if idx == f.numBlocks()-1 { + return f.plaintextLength - idx*int64(f.blockSize) + } + + return int64(f.blockSize) +} + +func (f *standardInputFile) physicalOffset(idx int64) int64 { Review Comment: small stdlib-conformance thing: a zero-length read on an empty file returns `io.EOF` here where `bytes.Reader`/`strings.Reader` return `(0, nil)`. With `plaintextLength` 0, `off >= f.plaintextLength` is `0 >= 0`, so `ReadAt(nil, 0)` reports `io.EOF`. A caller probing an empty file (a valid zero-row manifest) via a zero-length `ReadAt` would misread it as an error. I'd add `if len(p) == 0 { return 0, nil }` ahead of the offset check. ########## encryption/standard_manager.go: ########## @@ -0,0 +1,511 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package encryption + +import ( + "context" + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "io/fs" + + icebergio "github.com/apache/iceberg-go/io" +) + +// Defaults for [StandardEncryptionManager]. +const ( + // StandardDefaultDEKLength is the default length, in bytes, of the + // per-file data encryption key (DEK) generated for AES-256-GCM. + StandardDefaultDEKLength = 32 + + // StandardDefaultBlockSize is the default plaintext block size, in + // bytes, used to split a file into independently authenticated AES-GCM + // blocks. Blocks allow random access (Seek/ReadAt) without buffering or + // decrypting the whole file. + StandardDefaultBlockSize = 64 * 1024 +) + +// Sentinel errors returned by [StandardEncryptionManager]. +var ( + // ErrKeyIDRequired is returned by + // [StandardEncryptionManager.NewEncryptedOutputFile] when keyID is empty. + // StandardEncryptionManager always encrypts, so it requires a KEK to wrap + // the generated DEK; use [PlaintextEncryptionManager] for unencrypted + // tables instead of passing an empty keyID here. + ErrKeyIDRequired = errors.New("encryption: StandardEncryptionManager requires a non-empty keyID") + + // ErrKeyMetadataRequired is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when keyMetadata is + // empty. StandardEncryptionManager always decrypts, so it requires the + // per-file key metadata produced by [StandardEncryptionManager.NewEncryptedOutputFile]. + ErrKeyMetadataRequired = errors.New("encryption: StandardEncryptionManager requires non-empty key metadata") + + // ErrUnsupportedKeyMetadataVersion is returned when key metadata was + // produced by a newer, incompatible encoding version. + ErrUnsupportedKeyMetadataVersion = errors.New("encryption: unsupported key metadata version") + + // ErrInvalidBlockSize is returned when a configured or decoded block + // size is not positive. + ErrInvalidBlockSize = errors.New("encryption: block size must be positive") + + // ErrInvalidKeyMetadata is returned by + // [StandardEncryptionManager.NewDecryptedInputFile] when decoded key + // metadata fails basic sanity checks (e.g. a negative plaintext length + // or a nonce prefix of the wrong size). Key metadata is untrusted input + // on a crypto read path, so it is validated rather than trusted blindly. + ErrInvalidKeyMetadata = errors.New("encryption: invalid key metadata") +) + +// standardKeyMetadataVersion is the current encoding version written by +// [StandardEncryptionManager]. It is bumped whenever the on-disk layout of +// standardKeyMetadata or the block ciphertext format changes incompatibly. +const standardKeyMetadataVersion = 1 + +// standardKeyMetadata is the JSON-encoded structure stored as the opaque +// [EncryptionKeyMetadata] for files produced by [StandardEncryptionManager]. +type standardKeyMetadata struct { + Version int `json:"v"` + KeyID string `json:"key-id"` + WrappedKey []byte `json:"wrapped-key"` + NoncePrefix []byte `json:"nonce-prefix"` + BlockSize int `json:"block-size"` + PlaintextLength int64 `json:"plaintext-length"` +} + +// StandardEncryptionManager is a generic, format-agnostic [EncryptionManager] +// that provides envelope encryption for arbitrary files (e.g. manifests, +// manifest lists, Puffin statistics) using a [KeyManagementClient] to wrap +// and unwrap a fresh AES-256-GCM data encryption key (DEK) per file. +// +// Each file is split into fixed-size plaintext blocks, and each block is +// sealed independently with AES-GCM using a unique nonce (a per-file random +// prefix combined with the block index). This bounds memory usage and +// supports random access (Seek/ReadAt) on the decrypted file without +// buffering or decrypting more than the requested blocks. +// +// StandardEncryptionManager always encrypts and always decrypts: it fails +// closed, returning [ErrKeyIDRequired] or [ErrKeyMetadataRequired] rather +// than silently falling back to plaintext. Use [PlaintextEncryptionManager] +// for tables or files that are not encrypted. +type StandardEncryptionManager struct { + kms KeyManagementClient + dekLength int + blockSize int +} + +var _ EncryptionManager = (*StandardEncryptionManager)(nil) + +// StandardManagerOption configures a [StandardEncryptionManager] created by +// [NewStandardEncryptionManager]. +type StandardManagerOption func(*StandardEncryptionManager) + +// WithDEKLength overrides the default data encryption key length (in bytes). +// Valid AES key lengths are 16, 24, or 32 bytes. +func WithDEKLength(length int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.dekLength = length } +} + +// WithBlockSize overrides the default plaintext block size (in bytes) used +// to split files for independent block-level authentication. +func WithBlockSize(size int) StandardManagerOption { + return func(m *StandardEncryptionManager) { m.blockSize = size } +} + +// NewStandardEncryptionManager creates a [StandardEncryptionManager] backed +// by kms. kms must not be nil. +func NewStandardEncryptionManager(kms KeyManagementClient, opts ...StandardManagerOption) *StandardEncryptionManager { + m := &StandardEncryptionManager{ + kms: kms, + dekLength: StandardDefaultDEKLength, + blockSize: StandardDefaultBlockSize, + } + for _, opt := range opts { + opt(m) + } + + return m +} + +// NewEncryptedOutputFile creates a new AES-GCM block-encrypted output file. +// keyID identifies the KEK used to wrap the freshly generated per-file DEK, +// and must be non-empty; otherwise [ErrKeyIDRequired] is returned. +func (m *StandardEncryptionManager) NewEncryptedOutputFile(ctx context.Context, writer icebergio.FileWriter, keyID string) (EncryptedOutputFile, error) { + if keyID == "" { + return nil, ErrKeyIDRequired + } + if m.blockSize <= 0 { + return nil, fmt.Errorf("%w: got %d", ErrInvalidBlockSize, m.blockSize) + } + + var ( + plainDEK, wrappedDEK []byte + err error + ) + if m.kms.SupportsKeyGeneration() { + plainDEK, wrappedDEK, err = m.kms.GenerateKey(ctx, keyID, m.dekLength) + if err != nil { + return nil, fmt.Errorf("encryption: failed to generate DEK: %w", err) + } + } else { + plainDEK = make([]byte, m.dekLength) + if _, err = rand.Read(plainDEK); err != nil { + return nil, fmt.Errorf("encryption: failed to generate DEK: %w", err) + } + if wrappedDEK, err = m.kms.WrapKey(ctx, keyID, plainDEK); err != nil { + return nil, fmt.Errorf("encryption: failed to wrap DEK: %w", err) + } + } + + aead, err := newStandardAEAD(plainDEK) + if err != nil { + return nil, err + } + + noncePrefix := make([]byte, 4) + if _, err := rand.Read(noncePrefix); err != nil { + return nil, fmt.Errorf("encryption: failed to generate nonce prefix: %w", err) + } + + return &standardOutputFile{ + FileWriter: writer, + aead: aead, + noncePrefix: noncePrefix, + blockSize: m.blockSize, + keyID: keyID, + wrappedKey: wrappedDEK, + }, nil +} + +// NewDecryptedInputFile wraps file for transparent block-level AES-GCM +// decryption. keyMetadata must be the non-empty blob produced by +// [StandardEncryptionManager.NewEncryptedOutputFile]; otherwise +// [ErrKeyMetadataRequired] is returned. +func (m *StandardEncryptionManager) NewDecryptedInputFile(ctx context.Context, file icebergio.File, keyMetadata EncryptionKeyMetadata) (EncryptedInputFile, error) { + if len(keyMetadata) == 0 { + return nil, ErrKeyMetadataRequired + } + + var meta standardKeyMetadata + if err := json.Unmarshal(keyMetadata, &meta); err != nil { + return nil, fmt.Errorf("encryption: failed to decode key metadata: %w", err) + } + if meta.Version != standardKeyMetadataVersion { + return nil, fmt.Errorf("%w: %d", ErrUnsupportedKeyMetadataVersion, meta.Version) + } + if meta.BlockSize <= 0 { + return nil, fmt.Errorf("%w: block-size must be positive, got %d", ErrInvalidKeyMetadata, meta.BlockSize) + } + if meta.PlaintextLength < 0 { + return nil, fmt.Errorf("%w: plaintext-length must be non-negative, got %d", ErrInvalidKeyMetadata, meta.PlaintextLength) + } + if len(meta.NoncePrefix) != 4 { + return nil, fmt.Errorf("%w: nonce-prefix must be 4 bytes, got %d", ErrInvalidKeyMetadata, len(meta.NoncePrefix)) + } + + plainDEK, err := m.kms.UnwrapKey(ctx, meta.KeyID, meta.WrappedKey) + if err != nil { + return nil, fmt.Errorf("encryption: failed to unwrap DEK: %w", err) + } + + aead, err := newStandardAEAD(plainDEK) + if err != nil { + return nil, err + } + + return &standardInputFile{ + underlying: file, + aead: aead, + noncePrefix: meta.NoncePrefix, + blockSize: meta.BlockSize, + plaintextLength: meta.PlaintextLength, + keyMetadata: keyMetadata, + }, nil +} + +func newStandardAEAD(key []byte) (cipher.AEAD, error) { + block, err := aes.NewCipher(key) + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrInvalidKeyLength, err) + } + gcm, err := cipher.NewGCM(block) + if err != nil { + return nil, fmt.Errorf("encryption: failed to create GCM: %w", err) + } + + return gcm, nil +} + +// standardBlockNonce derives the AES-GCM nonce for blockIndex: the 4-byte +// per-file random prefix followed by the 8-byte big-endian block index. The +// (key, nonce) pair is unique per block since the DEK is fresh per file and +// no two blocks in the same file share an index. +func standardBlockNonce(prefix []byte, blockIndex uint64) []byte { + nonce := make([]byte, 12) + copy(nonce, prefix) + binary.BigEndian.PutUint64(nonce[4:], blockIndex) + + return nonce +} + +// standardOutputFile is an [EncryptedOutputFile] that seals fixed-size +// plaintext blocks with AES-GCM as they are written. +type standardOutputFile struct { + icebergio.FileWriter + + aead cipher.AEAD Review Comment: We bump `f.written` before the block is actually flushed, and I think that corrupts the file on a mid-write flush failure. If `flushBlock` fails, we reset `f.buf` but leave `f.written` counting the bytes that never reached the writer, so `Close` records a `PlaintextLength` longer than the ciphertext actually holds — the reader then computes wrong block boundaries and fails to auth the final block. The `return total - len(p)` on the same path also reports buffered-but-unflushed bytes as written, which an `io.Writer` caller treats as durable. I'd only advance `written` after a successful flush, and make the writer sticky (an `f.err`) so a failed flush poisons subsequent Writes rather than continuing on a half-written stream. While we're here, `Write` has no guard against use after `Close` — a Write after Close happily appends through the closed `FileWriter`. wdyt? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
