laskoviymishka commented on code in PR #1620: URL: https://github.com/apache/iceberg-go/pull/1620#discussion_r3850550091
########## encryption/standard_manager.go: ########## @@ -0,0 +1,583 @@ +// 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") + + // ErrOutputFileClosed is returned by [standardOutputFile.Write] when + // called after Close, or after a previous flush has poisoned the writer. + ErrOutputFileClosed = errors.New("encryption: write to closed StandardEncryptionManager output file") +) + +// 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) + } + switch m.dekLength { + case 16, 24, 32: + default: + return nil, fmt.Errorf("%w: DEK length must be 16, 24, or 32 bytes; got %d", ErrInvalidKeyLength, m.dekLength) + } + + // The (key, nonce) uniqueness this block format relies on requires a + // freshly generated DEK for every file: the nonce is only 4 random bytes + // plus a block index, so reusing a DEK across files would reuse (DEK, + // nonce) pairs and break AES-GCM's security guarantees. Never cache or + // reuse plainDEK/wrappedDEK across calls to NewEncryptedOutputFile. + 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) + } + } Review Comment: Small one: `crypto/rand.Read` always returns `(n, nil)` since Go 1.20, so this error check (and the one on the nonce prefix at ~205) is dead. `kms.go` in this package uses `io.ReadFull(rand.Reader, buf)`; matching that here keeps the check live and the crypto path consistent. ########## encryption/standard_manager.go: ########## @@ -0,0 +1,583 @@ +// 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") + + // ErrOutputFileClosed is returned by [standardOutputFile.Write] when + // called after Close, or after a previous flush has poisoned the writer. + ErrOutputFileClosed = errors.New("encryption: write to closed StandardEncryptionManager output file") +) + +// 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) + } + switch m.dekLength { + case 16, 24, 32: + default: + return nil, fmt.Errorf("%w: DEK length must be 16, 24, or 32 bytes; got %d", ErrInvalidKeyLength, m.dekLength) + } + + // The (key, nonce) uniqueness this block format relies on requires a + // freshly generated DEK for every file: the nonce is only 4 random bytes + // plus a block index, so reusing a DEK across files would reuse (DEK, + // nonce) pairs and break AES-GCM's security guarantees. Never cache or + // reuse plainDEK/wrappedDEK across calls to NewEncryptedOutputFile. + 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: %w", 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. +// Uniqueness of the (key, nonce) pair across every block ever sealed with a +// given DEK depends entirely on the DEK being freshly generated per file; +// see the caution in [StandardEncryptionManager.NewEncryptedOutputFile]. +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 + err error + + keyMetadata EncryptionKeyMetadata +} + +var _ EncryptedOutputFile = (*standardOutputFile)(nil) + +func (f *standardOutputFile) Write(p []byte) (int, error) { + if f.err != nil { + return 0, f.err + } + if f.closed { + return 0, ErrOutputFileClosed + } + + 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:] + if len(f.buf) == f.blockSize { + if err := f.flushBlock(); err != nil { + f.err = err + + return total - len(p), err + } + } + } + + return total, nil +} Review Comment: The count side of that same write concern. When a block flush fails mid-Write we return `total - len(p)`, but the failed block is still sitting in `f.buf`, drained out of `p` and never written. So `Write([]byte("aaaabbbb"))` with `blockSize=4` and the second flush failing returns `(8, err)` when only 4 bytes actually landed, and `io.Copy` accumulates that count before it checks the error. I'd return `total - len(p) - len(f.buf)` so we only count what was flushed. The regression test discards the count (`_, err = out.Write(...)`), which is why this slips through, so worth asserting `n` there too. ########## encryption/standard_manager.go: ########## @@ -0,0 +1,583 @@ +// 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") + + // ErrOutputFileClosed is returned by [standardOutputFile.Write] when + // called after Close, or after a previous flush has poisoned the writer. + ErrOutputFileClosed = errors.New("encryption: write to closed StandardEncryptionManager output file") +) + +// 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) + } + switch m.dekLength { + case 16, 24, 32: + default: + return nil, fmt.Errorf("%w: DEK length must be 16, 24, or 32 bytes; got %d", ErrInvalidKeyLength, m.dekLength) + } + + // The (key, nonce) uniqueness this block format relies on requires a + // freshly generated DEK for every file: the nonce is only 4 random bytes + // plus a block index, so reusing a DEK across files would reuse (DEK, + // nonce) pairs and break AES-GCM's security guarantees. Never cache or + // reuse plainDEK/wrappedDEK across calls to NewEncryptedOutputFile. + 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: %w", 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. +// Uniqueness of the (key, nonce) pair across every block ever sealed with a +// given DEK depends entirely on the DEK being freshly generated per file; +// see the caution in [StandardEncryptionManager.NewEncryptedOutputFile]. +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 + err error + + keyMetadata EncryptionKeyMetadata +} + +var _ EncryptedOutputFile = (*standardOutputFile)(nil) + +func (f *standardOutputFile) Write(p []byte) (int, error) { + if f.err != nil { + return 0, f.err + } + if f.closed { + return 0, ErrOutputFileClosed + } + + 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:] + if len(f.buf) == f.blockSize { + if err := f.flushBlock(); err != nil { + f.err = err + + return total - len(p), err + } + } + } + + return total, nil +} + +// flushBlock seals and writes the currently buffered plaintext block. +// f.written is only advanced once the ciphertext has actually reached the +// underlying writer, so a failed flush never overcounts PlaintextLength. +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.written += int64(len(f.buf)) + 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, max(32*1024, f.blockSize)) + 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 +} + +// Close flushes any buffered partial block and finalizes the key metadata. +// closed is only set once everything, including the underlying Close, has +// succeeded; a failed Close poisons the writer (via f.err) so a retry +// reliably reports the same error instead of masking the failure as success. +func (f *standardOutputFile) Close() error { + if f.err != nil { + return f.err + } + if f.closed { + return nil + } + + if len(f.buf) > 0 { + if err := f.flushBlock(); err != nil { + f.err = err + _ = f.FileWriter.Close() + + return err + } + } + + if f.keyMetadata == nil { + 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.err = fmt.Errorf("encryption: failed to encode key metadata: %w", err) + _ = f.FileWriter.Close() + + return f.err + } + f.keyMetadata = encoded + } + + if err := f.FileWriter.Close(); err != nil { + return fmt.Errorf("encryption: failed to close underlying writer: %w", err) + } + f.closed = true + + return nil +} Review Comment: This is the other half of the write/close note from last round, still open on the close path. We assign `f.keyMetadata` before calling `f.FileWriter.Close()`, and when that underlying Close fails we return the error but never set `f.err`. So a caller that reads `KeyMetadata()` after a failed Close gets a fully-populated blob for a file that never committed, and a retried Close sails past the `f.err` guard, calls the underlying Close a second time, and returns nil, masking the original failure. I'd set `f.err` on the underlying-close failure and clear `keyMetadata` (or gate `KeyMetadata()` on `f.closed`), so the failure sticks and the metadata only surfaces once the file's actually down. wdyt? ########## encryption/standard_manager.go: ########## @@ -0,0 +1,583 @@ +// 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") + + // ErrOutputFileClosed is returned by [standardOutputFile.Write] when + // called after Close, or after a previous flush has poisoned the writer. + ErrOutputFileClosed = errors.New("encryption: write to closed StandardEncryptionManager output file") +) + +// 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) + } + switch m.dekLength { + case 16, 24, 32: + default: + return nil, fmt.Errorf("%w: DEK length must be 16, 24, or 32 bytes; got %d", ErrInvalidKeyLength, m.dekLength) + } + + // The (key, nonce) uniqueness this block format relies on requires a + // freshly generated DEK for every file: the nonce is only 4 random bytes + // plus a block index, so reusing a DEK across files would reuse (DEK, + // nonce) pairs and break AES-GCM's security guarantees. Never cache or + // reuse plainDEK/wrappedDEK across calls to NewEncryptedOutputFile. + 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 { Review Comment: `block-size` comes off untrusted JSON and we only check `<= 0`. With no upper bound it feeds two bad spots: `physicalOffset` does `f.blockSize + f.aead.Overhead()` as an int add before the int64 cast, so a large decoded block-size overflows to a negative offset we then hand to `ReadAt`; and `readBlock` sizes `make([]byte, wantLen)` off it, so a 1 GiB block-size means a 1 GiB allocation per block read, an OOM any malformed manifest entry can trigger. I'd add an upper bound here (a few tens of MiB) and do the offset add in int64. Fails closed the way the rest of the read path does now. -- 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]
