You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2026/10/14 13:23:57 UTC
(iceberg-go) branch main updated: feat(encryption): add StandardEncryptionManager for generic file envelope encryption (#1620)
This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 0fc66726e feat(encryption): add StandardEncryptionManager for generic file envelope encryption (#1620)
0fc66726e is described below
commit 0fc66726eab8eeb424449be48983ee6614c729a9
Author: Hectar <84...@users.noreply.github.com>
AuthorDate: Wed Oct 14 06:23:50 2026 -0700
feat(encryption): add StandardEncryptionManager for generic file envelope encryption (#1620)
* feat(encryption): add StandardEncryptionManager for generic file envelope encryption.
Signed-off-by: hectar-glitches <he...@uni.minerva.edu>
* fix(encryption): validate block size and decoded key metadata in StandardEncryptionManager.
Signed-off-by: hectar-glitches <he...@uni.minerva.edu>
* fix(encryption): harden standard manager I/O
Signed-off-by: hectar-glitches <he...@uni.minerva.edu>
* Address blockers
* fix major concerns with filewriter and standardinputfile
* address the major and minor issues, rename the identifier from standard to go envelope
* address memory limit issue and nits
* fix(encryption): add nonce-uniqueness test, bound aad-prefix length
* chore(encryption): rebase onto main, fix modernize lint
* feat(encryption): make StandardEncryptionManager Java-compatible
Read and write the same bytes as Java's StandardEncryptionManager and
iceberg-rust, so files can be exchanged between implementations.
- Use a fixed 1 MiB plaintext block size. Remove WithBlockSize,
GoEnvelopeMaxBlockSize and ErrInvalidBlockSize. The reader requires
the stream header to declare 1 MiB.
- Encode key metadata as StandardKeyMetadata: a version byte followed by
an Avro record {encryption_key, aad_prefix, file_length}. The DEK is
generated locally and stored raw; there is no KMS call per file. The
reader derives the plaintext length from file_length.
- Write block 0 for an empty file, giving the 36-byte layout Java and
iceberg-rust use, and accept it on read.
- Default the data key length to 16 bytes and rename the option to
WithStandardDataKeyLength.
- Rename GoEnvelopeEncryptionManager back to StandardEncryptionManager.
- Document that ErrOutputFileClosed is only returned after a successful
Close.
Tests decode the writer's output with a stdlib-only AGS1 decoder, read a
stream built the same way, and check a hand-encoded key metadata vector.
Block-size tests now use multi-MiB payloads.
* fixes to PR description, major java issue and minor issues
---------
Signed-off-by: hectar-glitches <he...@uni.minerva.edu>
Co-authored-by: Matt Topol <zo...@gmail.com>
---
encryption/standard_manager.go | 888 +++++++++++++++++++++++++++++++++
encryption/standard_manager_test.go | 947 ++++++++++++++++++++++++++++++++++++
2 files changed, 1835 insertions(+)
diff --git a/encryption/standard_manager.go b/encryption/standard_manager.go
new file mode 100644
index 000000000..fdac33de6
--- /dev/null
+++ b/encryption/standard_manager.go
@@ -0,0 +1,888 @@
+// 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"
+ "errors"
+ "fmt"
+ "io"
+ "io/fs"
+ "math"
+ "slices"
+ "sync"
+
+ icebergio "github.com/apache/iceberg-go/io"
+)
+
+// StandardDefaultDataKeyLength is the default length, in bytes, of the
+// per-file data encryption key (DEK). It is 128-bit, matching the default of
+// the encryption.data-key-length table property in the Java and Rust
+// implementations.
+const StandardDefaultDataKeyLength = 16
+
+// Sentinel errors returned by [StandardEncryptionManager].
+var (
+ // ErrKeyIDRequired is returned by
+ // [StandardEncryptionManager.NewEncryptedOutputFile] when keyID is empty.
+ // StandardEncryptionManager always encrypts, so it requires the table's
+ // encryption key ID; 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] and
+ // [StandardEncryptionManager.NewDecryptedInputFileWithLength] 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 starts
+ // with a version byte other than the one this package understands.
+ ErrUnsupportedKeyMetadataVersion = errors.New("encryption: unsupported key metadata version")
+
+ // ErrInvalidStreamHeader is returned when the AES GCM Stream header (the
+ // "AGS1" magic and little-endian plain block length at the start of an
+ // encrypted file, per the Iceberg AES GCM Stream spec) is missing,
+ // truncated, or declares a block length other than the fixed 1 MiB.
+ ErrInvalidStreamHeader = errors.New("encryption: invalid AES GCM Stream header")
+
+ // ErrInvalidKeyMetadata is returned by
+ // [StandardEncryptionManager.NewDecryptedInputFile] and
+ // [StandardEncryptionManager.NewDecryptedInputFileWithLength] when key
+ // metadata is not a well-formed StandardKeyMetadata record or fails basic
+ // sanity checks (e.g. a truncated record, a missing or wrong-sized AAD
+ // prefix, or a missing or impossible encrypted length). 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 a successful Close. It wraps [fs.ErrClosed] so callers can
+ // test with errors.Is(err, fs.ErrClosed). A Write after a failed flush or
+ // a failed Close returns that original error instead.
+ ErrOutputFileClosed = fmt.Errorf("encryption: write to closed StandardEncryptionManager output file: %w", fs.ErrClosed)
+
+ // ErrBlockTruncated is returned by [standardInputFile.ReadAt] when the
+ // underlying storage returns fewer ciphertext bytes for a block than its
+ // recorded length requires, indicating the file was truncated at rest.
+ // This is distinct from [ErrCiphertextTooShort], which [KeyManagementClient]
+ // implementations use for a too-short wrapped key or KMS-encrypted
+ // payload; keeping them separate lets a caller tell a malformed KMS blob
+ // apart from a short block read.
+ ErrBlockTruncated = errors.New("encryption: block truncated: read fewer ciphertext bytes than expected")
+)
+
+// Constants describing the Iceberg AES GCM Stream ("AGS1") wire format used
+// for the ciphertext produced by [StandardEncryptionManager]. See
+// https://iceberg.apache.org/gcm-stream-spec/ for the full specification.
+const (
+ // gcmStreamMagic identifies an AES GCM Stream version 1 file.
+ gcmStreamMagic = "AGS1"
+
+ // gcmStreamHeaderLength is the length, in bytes, of the magic plus the
+ // 4-byte little-endian plain block length written at the start of every
+ // file.
+ gcmStreamHeaderLength = 8
+
+ // gcmStreamPlainBlockSize is the plaintext block size. It is fixed:
+ // Java's Ciphers.PLAIN_BLOCK_SIZE and iceberg-rust's PLAIN_BLOCK_SIZE are
+ // both 1 MiB, and Java's AesGcmInputStream rejects any other value in the
+ // header, so a stream with a different block size is unreadable there.
+ gcmStreamPlainBlockSize = 1024 * 1024
+
+ // gcmStreamNonceLength is the length, in bytes, of the random AES-GCM
+ // nonce stored at the start of every cipher block.
+ gcmStreamNonceLength = 12
+
+ // gcmStreamTagLength is the length, in bytes, of the AES-GCM
+ // authentication tag appended to every cipher block's ciphertext.
+ gcmStreamTagLength = 16
+
+ // gcmStreamBlockOverhead is the number of ciphertext bytes added to
+ // each block beyond its plaintext length (nonce + tag).
+ gcmStreamBlockOverhead = gcmStreamNonceLength + gcmStreamTagLength
+
+ // gcmStreamCipherBlockSize is the on-disk length of a full block.
+ gcmStreamCipherBlockSize = gcmStreamPlainBlockSize + gcmStreamBlockOverhead
+
+ // gcmStreamMinLength is the length of the smallest valid stream: the
+ // header followed by one block holding no plaintext. Java and iceberg-rust
+ // always write block 0, even for an empty file, and iceberg-rust rejects
+ // anything shorter.
+ gcmStreamMinLength = gcmStreamHeaderLength + gcmStreamBlockOverhead
+
+ // gcmStreamAADPrefixLength is the length, in bytes, of the random
+ // per-file AAD prefix. It matches Java's ENCRYPTION_AAD_LENGTH_DEFAULT.
+ gcmStreamAADPrefixLength = 16
+)
+
+// standardKeyMetadataVersion is the leading version byte of Iceberg's
+// StandardKeyMetadata encoding.
+const standardKeyMetadataVersion = 1
+
+// standardKeyMetadata is the decoded form of Iceberg's StandardKeyMetadata:
+// a version byte followed by an Avro binary record
+//
+// {encryption_key: bytes, aad_prefix: ["null", bytes], file_length: ["null", long]}
+//
+// This is the key_metadata encoding Java's StandardEncryptionManager,
+// iceberg-rust and pyiceberg read and write. It holds the raw DEK, so it is a
+// secret and must be protected by whatever stores it.
+type standardKeyMetadata struct {
+ encryptionKey []byte
+
+ // aadPrefix is combined with each block's little-endian index to form
+ // the AES GCM Stream additional authenticated data, binding every
+ // ciphertext block to this file and to its position so that blocks
+ // cannot be silently reordered, replayed from another file, or spliced
+ // in from elsewhere in the same file. It is not secret. It is nil when
+ // the record omits it.
+ aadPrefix []byte
+
+ // fileLength is the length of the encrypted stream, header included.
+ // The reader derives the plaintext length from it, per the AES GCM
+ // Stream spec's "File length" note. hasFileLength is false when the
+ // record omits it.
+ fileLength int64
+ hasFileLength bool
+}
+
+// encodeStandardKeyMetadata encodes the version byte and Avro record for a
+// file, with both optional fields present.
+func encodeStandardKeyMetadata(encryptionKey, aadPrefix []byte, fileLength int64) EncryptionKeyMetadata {
+ b := []byte{standardKeyMetadataVersion}
+ b = binary.AppendVarint(b, int64(len(encryptionKey)))
+ b = append(b, encryptionKey...)
+ b = binary.AppendVarint(b, 1) // union branch 1 (bytes): aad_prefix is present
+ b = binary.AppendVarint(b, int64(len(aadPrefix)))
+ b = append(b, aadPrefix...)
+ b = binary.AppendVarint(b, 1) // union branch 1 (long): file_length is present
+ b = binary.AppendVarint(b, fileLength)
+
+ return b
+}
+
+// decodeStandardKeyMetadata decodes the version byte and Avro record written
+// by [encodeStandardKeyMetadata] or by Java's StandardKeyMetadata.buffer().
+func decodeStandardKeyMetadata(b []byte) (standardKeyMetadata, error) {
+ if b[0] != standardKeyMetadataVersion {
+ return standardKeyMetadata{}, fmt.Errorf("%w: %d", ErrUnsupportedKeyMetadataVersion, b[0])
+ }
+
+ r := avroReader{buf: b[1:]}
+ var k standardKeyMetadata
+ k.encryptionKey = r.bytes()
+ if r.present() {
+ k.aadPrefix = r.bytes()
+ }
+ if r.present() {
+ k.fileLength = r.long()
+ k.hasFileLength = true
+ }
+ if r.err == nil && len(r.buf) != 0 {
+ r.err = fmt.Errorf("%d unexpected trailing bytes", len(r.buf))
+ }
+ if r.err != nil {
+ return standardKeyMetadata{}, fmt.Errorf("%w: %w", ErrInvalidKeyMetadata, r.err)
+ }
+
+ return k, nil
+}
+
+// avroReader decodes the few Avro binary primitives StandardKeyMetadata uses,
+// remembering the first error. Avro longs are zig-zag varints, which is what
+// encoding/binary's Varint implements.
+type avroReader struct {
+ buf []byte
+ err error
+}
+
+func (r *avroReader) long() int64 {
+ if r.err != nil {
+ return 0
+ }
+ v, n := binary.Varint(r.buf)
+ if n <= 0 {
+ r.err = errors.New("truncated or oversized varint")
+
+ return 0
+ }
+ r.buf = r.buf[n:]
+
+ return v
+}
+
+func (r *avroReader) bytes() []byte {
+ n := r.long()
+ if r.err != nil {
+ return nil
+ }
+ if n < 0 || n > int64(len(r.buf)) {
+ r.err = fmt.Errorf("bytes length %d out of range (%d bytes remain)", n, len(r.buf))
+
+ return nil
+ }
+ out := slices.Clone(r.buf[:n])
+ r.buf = r.buf[n:]
+
+ return out
+}
+
+// present reads the branch index of a ["null", T] union and reports whether
+// the value is present (branch 1) rather than null (branch 0).
+func (r *avroReader) present() bool {
+ branch := r.long()
+ if r.err != nil {
+ return false
+ }
+ if branch != 0 && branch != 1 {
+ r.err = fmt.Errorf("unexpected union branch %d", branch)
+
+ return false
+ }
+
+ return branch == 1
+}
+
+// plaintextLengthFromStreamLength derives the plaintext length from the
+// length of the encrypted stream, the way Java's
+// AesGcmInputStream.calculatePlaintextLength does. It accepts the 36-byte
+// form of an empty file (header plus one empty block) and rejects lengths no
+// writer could have produced.
+func plaintextLengthFromStreamLength(streamLength int64) (int64, error) {
+ if streamLength < gcmStreamMinLength {
+ return 0, fmt.Errorf("%w: file-length %d is shorter than the %d-byte minimum stream", ErrInvalidKeyMetadata, streamLength, gcmStreamMinLength)
+ }
+
+ body := streamLength - gcmStreamHeaderLength
+ plainLength := body / gcmStreamCipherBlockSize * gcmStreamPlainBlockSize
+ if lastBlock := body % gcmStreamCipherBlockSize; lastBlock != 0 {
+ if lastBlock < gcmStreamBlockOverhead {
+ return 0, fmt.Errorf("%w: file-length %d ends in a %d-byte block, shorter than its %d-byte nonce and tag", ErrInvalidKeyMetadata, streamLength, lastBlock, gcmStreamBlockOverhead)
+ }
+ plainLength += lastBlock - gcmStreamBlockOverhead
+ }
+
+ return plainLength, nil
+}
+
+// StandardEncryptionManager is a generic, format-agnostic [EncryptionManager]
+// for arbitrary files (e.g. manifests, manifest lists, Puffin statistics). It
+// reads and writes the same bytes as Java's StandardEncryptionManager,
+// iceberg-rust and pyiceberg: an Iceberg AES GCM Stream ("AGS1") with fixed
+// 1 MiB plaintext blocks, and key metadata in Iceberg's StandardKeyMetadata
+// encoding.
+//
+// Each file is encrypted under a fresh random data encryption key (DEK),
+// generated locally with no KMS call. Each block carries its own random nonce
+// and is authenticated with an AAD that binds it to the file and to its
+// position. This bounds memory usage and supports random access
+// (Seek/ReadAt) on the decrypted file without buffering or decrypting more
+// than the requested blocks.
+//
+// The key metadata returned by Close contains the raw DEK, so it is a secret:
+// persist it only where something protects it, such as a manifest or manifest
+// list that is itself encrypted under a key encryption key.
+//
+// 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 {
+ dataKeyLength int
+}
+
+var _ EncryptionManager = (*StandardEncryptionManager)(nil)
+
+// StandardManagerOption configures a [StandardEncryptionManager] created by
+// [NewStandardEncryptionManager].
+type StandardManagerOption func(*StandardEncryptionManager)
+
+// WithStandardDataKeyLength sets the length, in bytes, of the per-file data
+// encryption key. It corresponds to the encryption.data-key-length table
+// property. Valid AES key lengths are 16, 24, or 32 bytes; the default is
+// [StandardDefaultDataKeyLength].
+func WithStandardDataKeyLength(length int) StandardManagerOption {
+ return func(m *StandardEncryptionManager) { m.dataKeyLength = length }
+}
+
+// NewStandardEncryptionManager creates a [StandardEncryptionManager].
+func NewStandardEncryptionManager(opts ...StandardManagerOption) *StandardEncryptionManager {
+ m := &StandardEncryptionManager{dataKeyLength: StandardDefaultDataKeyLength}
+ for _, opt := range opts {
+ opt(m)
+ }
+
+ return m
+}
+
+// NewEncryptedOutputFile creates a new AES-GCM block-encrypted output file.
+// keyID is the table's encryption key ID and must be non-empty; otherwise
+// [ErrKeyIDRequired] is returned. keyID is only checked: the file's key
+// metadata holds the raw DEK.
+func (m *StandardEncryptionManager) NewEncryptedOutputFile(_ context.Context, writer icebergio.FileWriter, keyID string) (EncryptedOutputFile, error) {
+ if keyID == "" {
+ return nil, ErrKeyIDRequired
+ }
+ switch m.dataKeyLength {
+ case 16, 24, 32:
+ default:
+ return nil, fmt.Errorf("%w: DEK length must be 16, 24, or 32 bytes; got %d", ErrInvalidKeyLength, m.dataKeyLength)
+ }
+
+ // A fresh DEK per file keeps the number of random-nonce encryptions under
+ // any one key within the 2^32 limit NIST SP 800-38D sets for random
+ // 96-bit nonces: a file holds at most 2^32-1 blocks.
+ dek := make([]byte, m.dataKeyLength)
+ if _, err := io.ReadFull(rand.Reader, dek); err != nil {
+ return nil, fmt.Errorf("encryption: failed to generate DEK: %w", err)
+ }
+
+ aead, err := newStandardAEAD(dek)
+ if err != nil {
+ return nil, err
+ }
+
+ aadPrefix := make([]byte, gcmStreamAADPrefixLength)
+ if _, err := io.ReadFull(rand.Reader, aadPrefix); err != nil {
+ return nil, fmt.Errorf("encryption: failed to generate AAD prefix: %w", err)
+ }
+
+ // Write the AES GCM Stream header (magic + little-endian block length)
+ // up front, before any ciphertext blocks, per the format spec.
+ header := make([]byte, gcmStreamHeaderLength)
+ copy(header, gcmStreamMagic)
+ binary.LittleEndian.PutUint32(header[len(gcmStreamMagic):], gcmStreamPlainBlockSize)
+ if _, err := writer.Write(header); err != nil {
+ _ = writer.Close()
+
+ return nil, fmt.Errorf("encryption: failed to write stream header: %w", err)
+ }
+
+ return &standardOutputFile{
+ FileWriter: writer,
+ aead: aead,
+ dek: dek,
+ aadPrefix: aadPrefix,
+ buf: make([]byte, 0, gcmStreamPlainBlockSize),
+ sealBuf: make([]byte, 0, gcmStreamCipherBlockSize),
+ encryptedLength: gcmStreamHeaderLength,
+ }, nil
+}
+
+// NewDecryptedInputFile wraps file for transparent block-level AES-GCM
+// decryption. keyMetadata must be a non-empty StandardKeyMetadata record, as
+// produced by [StandardEncryptionManager.NewEncryptedOutputFile] or another
+// Iceberg implementation; otherwise [ErrKeyMetadataRequired] is returned. The
+// record must carry a 16-byte AAD prefix and a file length.
+//
+// The plaintext length is derived from that file length. For key metadata
+// without one, use [StandardEncryptionManager.NewDecryptedInputFileWithLength].
+func (m *StandardEncryptionManager) NewDecryptedInputFile(ctx context.Context, file icebergio.File, keyMetadata EncryptionKeyMetadata) (EncryptedInputFile, error) {
+ return m.newDecryptedInputFile(ctx, file, -1, keyMetadata)
+}
+
+// NewDecryptedInputFileWithLength is like
+// [StandardEncryptionManager.NewDecryptedInputFile], but accepts key metadata
+// without a file length, as Java writes for data files and manifests.
+// encryptedLength is the length of the encrypted file, header included, as
+// recorded in table metadata (DataFile.FileSizeBytes, ManifestFile.Length). A
+// file length in the key metadata takes precedence.
+func (m *StandardEncryptionManager) NewDecryptedInputFileWithLength(ctx context.Context, file icebergio.File, encryptedLength int64, keyMetadata EncryptionKeyMetadata) (EncryptedInputFile, error) {
+ if encryptedLength < 0 {
+ return nil, fmt.Errorf("%w: encrypted length must be non-negative, got %d", ErrInvalidKeyMetadata, encryptedLength)
+ }
+
+ return m.newDecryptedInputFile(ctx, file, encryptedLength, keyMetadata)
+}
+
+// newDecryptedInputFile implements both read entry points. callerLength is the
+// caller-supplied encrypted length, or -1 when there is none.
+func (m *StandardEncryptionManager) newDecryptedInputFile(_ context.Context, file icebergio.File, callerLength int64, keyMetadata EncryptionKeyMetadata) (EncryptedInputFile, error) {
+ if len(keyMetadata) == 0 {
+ return nil, ErrKeyMetadataRequired
+ }
+
+ meta, err := decodeStandardKeyMetadata(keyMetadata)
+ if err != nil {
+ return nil, err
+ }
+ // Java, iceberg-rust and this writer all emit 16-byte prefixes. Bounding
+ // it also keeps the per-block AAD allocation in readBlock small.
+ if len(meta.aadPrefix) != gcmStreamAADPrefixLength {
+ return nil, fmt.Errorf("%w: aad-prefix must be exactly %d bytes, got %d", ErrInvalidKeyMetadata, gcmStreamAADPrefixLength, len(meta.aadPrefix))
+ }
+ streamLength := callerLength
+ switch {
+ case meta.hasFileLength:
+ streamLength = meta.fileLength
+ case callerLength < 0:
+ return nil, fmt.Errorf("%w: file-length is required when no encrypted length is supplied", ErrInvalidKeyMetadata)
+ }
+ plaintextLength, err := plaintextLengthFromStreamLength(streamLength)
+ if err != nil {
+ return nil, err
+ }
+
+ aead, err := newStandardAEAD(meta.encryptionKey)
+ if err != nil {
+ return nil, err
+ }
+
+ if err := validateGCMStreamHeader(file); err != nil {
+ return nil, err
+ }
+
+ return &standardInputFile{
+ underlying: file,
+ aead: aead,
+ aadPrefix: meta.aadPrefix,
+ plaintextLength: 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
+}
+
+// validateGCMStreamHeader reads the AES GCM Stream header at the start of
+// file and checks that it is the "AGS1" magic followed by the fixed 1 MiB
+// plain block length. Reads use [gcmStreamPlainBlockSize] as the block size.
+func validateGCMStreamHeader(file icebergio.File) error {
+ header := make([]byte, gcmStreamHeaderLength)
+ n, err := file.ReadAt(header, 0)
+ if err != nil && !errors.Is(err, io.EOF) {
+ return fmt.Errorf("encryption: failed to read stream header: %w", err)
+ }
+ if n != gcmStreamHeaderLength {
+ return fmt.Errorf("%w: expected %d header bytes, got %d", ErrInvalidStreamHeader, gcmStreamHeaderLength, n)
+ }
+ if string(header[:len(gcmStreamMagic)]) != gcmStreamMagic {
+ return fmt.Errorf("%w: missing %q magic", ErrInvalidStreamHeader, gcmStreamMagic)
+ }
+
+ if blockSize := binary.LittleEndian.Uint32(header[len(gcmStreamMagic):]); blockSize != gcmStreamPlainBlockSize {
+ return fmt.Errorf("%w: block length %d, want %d", ErrInvalidStreamHeader, blockSize, gcmStreamPlainBlockSize)
+ }
+
+ return nil
+}
+
+// gcmStreamBlockAAD derives the AES-GCM additional authenticated data for
+// blockIndex: the per-file AAD prefix followed by the 4-byte little-endian
+// block index, per the Iceberg AES GCM Stream spec. GCM authenticates the AAD
+// along with the ciphertext, so a block only verifies at its own index in its
+// own file: reordered, duplicated, or spliced-in blocks fail authentication.
+func gcmStreamBlockAAD(prefix []byte, blockIndex uint32) []byte {
+ aad := make([]byte, len(prefix)+4)
+ copy(aad, prefix)
+ binary.LittleEndian.PutUint32(aad[len(prefix):], blockIndex)
+
+ return aad
+}
+
+// standardOutputFile is an [EncryptedOutputFile] that seals fixed-size
+// plaintext blocks with AES-GCM as they are written, using the Iceberg AES
+// GCM Stream ("AGS1") wire format.
+type standardOutputFile struct {
+ icebergio.FileWriter
+
+ aead cipher.AEAD
+ dek []byte
+ aadPrefix []byte
+
+ // buf holds the pending plaintext block and sealBuf is scratch for its
+ // nonce, ciphertext and tag. Each is allocated once at its full-block
+ // size and reused for every block.
+ buf []byte
+ sealBuf []byte
+ blockIndex uint32
+ closed bool
+ err error
+
+ // encryptedLength is the number of bytes written to the underlying
+ // writer so far, header included. It becomes file_length in the key
+ // metadata, and only advances once a block has been written.
+ encryptedLength int64
+
+ // underlyingClosed records that FileWriter.Close has been called, so the
+ // underlying writer is closed at most once.
+ underlyingClosed bool
+
+ keyMetadata EncryptionKeyMetadata
+}
+
+var _ EncryptedOutputFile = (*standardOutputFile)(nil)
+
+// closeUnderlyingIgnoringError closes the underlying writer at most once.
+// The error is ignored: the caller is already reporting a more specific
+// failure (a flush error), and this is best-effort cleanup so a poisoned
+// writer never leaks its underlying file descriptor or connection.
+func (f *standardOutputFile) closeUnderlyingIgnoringError() {
+ if f.underlyingClosed {
+ return
+ }
+ f.underlyingClosed = true
+ _ = f.FileWriter.Close()
+}
+
+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)
+ consumed := 0 // bytes of p appended into f.buf so far in this call
+ accepted := 0 // bytes of p known to be durably flushed; reported on failure
+ for len(p) > 0 {
+ space := gcmStreamPlainBlockSize - len(f.buf)
+ n := min(space, len(p))
+ f.buf = append(f.buf, p[:n]...)
+ p = p[n:]
+ consumed += n
+ if len(f.buf) == gcmStreamPlainBlockSize {
+ if err := f.flushBlock(); err != nil {
+ f.err = err
+ f.closeUnderlyingIgnoringError()
+
+ return accepted, err
+ }
+ accepted = consumed
+ }
+ }
+
+ return total, nil
+}
+
+// flushBlock seals and writes the currently buffered plaintext block using a
+// fresh random nonce, per the Iceberg AES GCM Stream format.
+// f.encryptedLength only advances once the ciphertext has reached the
+// underlying writer, so file_length never counts a block that failed to write.
+func (f *standardOutputFile) flushBlock() error {
+ if f.blockIndex == math.MaxUint32 {
+ return errors.New("encryption: cannot write block: exceeded maximum block count")
+ }
+
+ nonce := f.sealBuf[:gcmStreamNonceLength]
+ if _, err := io.ReadFull(rand.Reader, nonce); err != nil {
+ return fmt.Errorf("encryption: failed to generate block nonce: %w", err)
+ }
+
+ aad := gcmStreamBlockAAD(f.aadPrefix, f.blockIndex)
+ sealed := f.aead.Seal(nonce, nonce, f.buf, aad)
+ if _, err := f.FileWriter.Write(sealed); err != nil {
+ return fmt.Errorf("encryption: failed to write encrypted block: %w", err)
+ }
+ f.encryptedLength += int64(len(sealed))
+ 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, gcmStreamPlainBlockSize)
+ 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, writes an empty block 0 if the
+// file is empty, and finalizes the key metadata. closed is set, and
+// keyMetadata published, only once everything, including the underlying
+// Close, has succeeded. A failed Close poisons the writer (via f.err): a retry
+// returns the same error, and no key metadata is exposed for an output that
+// never finished.
+func (f *standardOutputFile) Close() error {
+ if f.err != nil {
+ return f.err
+ }
+ if f.closed {
+ return nil
+ }
+
+ // Java and iceberg-rust write block 0 even when the file is empty, so an
+ // empty file is 36 bytes (a header plus one 28-byte block), and
+ // iceberg-rust rejects anything shorter. Match them.
+ if len(f.buf) > 0 || f.blockIndex == 0 {
+ if err := f.flushBlock(); err != nil {
+ f.err = err
+ f.closeUnderlyingIgnoringError()
+
+ return err
+ }
+ }
+
+ if err := f.FileWriter.Close(); err != nil {
+ f.underlyingClosed = true
+ f.err = fmt.Errorf("encryption: failed to close underlying writer: %w", err)
+
+ return f.err
+ }
+ f.underlyingClosed = true
+
+ f.keyMetadata = encodeStandardKeyMetadata(f.dek, f.aadPrefix, f.encryptedLength)
+ f.closed = true
+
+ return nil
+}
+
+// KeyMetadata returns the finalized per-file key metadata. It is only
+// populated after Close has succeeded. It contains the raw DEK.
+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.
+//
+// The plaintext length is derived from the encrypted length (the file length
+// in key metadata, or the caller's), so ciphertext appended after the last
+// block is ignored. Both follow
+// from the AES GCM Stream spec's "File length" rule.
+//
+// ReadAt is safe, and intended, for concurrent use, matching the io.ReaderAt
+// contract: readBlock only holds cacheMu around the cache check/publish, not
+// across the underlying ReadAt or AEAD decryption, so concurrent calls for
+// distinct blocks run in parallel. A lost cache race (two goroutines
+// decrypting the same block) is harmless, just wasted work. Read and Seek
+// mutate the shared cursor (pos) and are not concurrent-safe; do not call them
+// from multiple goroutines on the same instance.
+//
+// The single-entry cache retains one decrypted block (1 MiB) for the lifetime
+// of the input file, not just the lifetime of a single read call.
+type standardInputFile struct {
+ underlying icebergio.File
+ aead cipher.AEAD
+ aadPrefix []byte
+ plaintextLength int64
+ keyMetadata EncryptionKeyMetadata
+
+ pos int64
+
+ // cacheMu guards cacheIdx/cacheBlock/cacheValid below. It is held only
+ // around the cache check and the cache publish in readBlock.
+ cacheMu sync.Mutex
+ cacheIdx int64
+ cacheBlock []byte
+ cacheValid bool
+}
+
+var _ EncryptedInputFile = (*standardInputFile)(nil)
+
+func (f *standardInputFile) numBlocks() int64 {
+ if f.plaintextLength == 0 {
+ return 0
+ }
+
+ // plaintextLength comes from key metadata or the caller and can be close
+ // to math.MaxInt64; 1 + (n-1)/size avoids the overflow that
+ // (n + size - 1)/size would hit there.
+ return 1 + (f.plaintextLength-1)/gcmStreamPlainBlockSize
+}
+
+// blockPlainLen returns the plaintext length of block idx, given the total
+// number of blocks.
+func (f *standardInputFile) blockPlainLen(idx, numBlocks int64) int64 {
+ if idx == numBlocks-1 {
+ return f.plaintextLength - idx*gcmStreamPlainBlockSize
+ }
+
+ return gcmStreamPlainBlockSize
+}
+
+// readBlock decrypts block idx, or returns it from the single-entry cache
+// if it was the most recently decrypted block. It checks the byte count
+// returned by ReadAt: a short read is reported as [ErrBlockTruncated]
+// (truncated storage), since zero-padding it into the AEAD would surface as a
+// misleading [ErrAuthenticationFailed].
+//
+// cacheMu is held only around the cache check and the cache publish at the
+// end, so concurrent readers do their I/O and decryption in parallel. Two
+// goroutines may decrypt the same block; the last one to finish publishes it.
+func (f *standardInputFile) readBlock(idx int64) ([]byte, error) {
+ f.cacheMu.Lock()
+ if f.cacheValid && f.cacheIdx == idx {
+ block := f.cacheBlock
+ f.cacheMu.Unlock()
+
+ return block, nil
+ }
+ f.cacheMu.Unlock()
+
+ numBlocks := f.numBlocks()
+ if idx < 0 || idx >= numBlocks {
+ return nil, fmt.Errorf("%w: block index %d out of range [0, %d)", ErrInvalidKeyMetadata, idx, numBlocks)
+ }
+ // The AAD carries the block index as 32 bits, and an encrypted length
+ // near math.MaxInt64 would otherwise imply more blocks than that.
+ if idx > math.MaxUint32 {
+ return nil, fmt.Errorf("%w: block index %d exceeds the maximum supported block count", ErrInvalidKeyMetadata, idx)
+ }
+
+ // idx < numBlocks and plaintextLength was derived from an int64 stream
+ // length, so this offset is no larger than that length and cannot
+ // overflow.
+ offset := gcmStreamHeaderLength + idx*gcmStreamCipherBlockSize
+
+ wantLen := f.blockPlainLen(idx, numBlocks) + gcmStreamBlockOverhead
+ ciphertext := make([]byte, wantLen)
+ n, err := f.underlying.ReadAt(ciphertext, offset)
+ if err != nil && !errors.Is(err, io.EOF) {
+ return nil, fmt.Errorf("encryption: failed to read block %d: %w", idx, err)
+ }
+ if int64(n) != wantLen {
+ return nil, fmt.Errorf("%w: block %d: read %d of %d expected ciphertext bytes", ErrBlockTruncated, idx, n, wantLen)
+ }
+
+ nonce, sealed := ciphertext[:gcmStreamNonceLength], ciphertext[gcmStreamNonceLength:]
+ aad := gcmStreamBlockAAD(f.aadPrefix, uint32(idx))
+
+ // Decrypt in place into this block's ciphertext buffer.
+ plaintext, err := f.aead.Open(sealed[:0], nonce, sealed, aad)
+ if err != nil {
+ return nil, fmt.Errorf("%w: block %d: %w", ErrAuthenticationFailed, idx, err)
+ }
+
+ f.cacheMu.Lock()
+ f.cacheIdx = idx
+ f.cacheBlock = plaintext
+ f.cacheValid = true
+ f.cacheMu.Unlock()
+
+ return plaintext, nil
+}
+
+func (f *standardInputFile) ReadAt(p []byte, off int64) (int, error) {
+ if len(p) == 0 {
+ return 0, nil
+ }
+ if off < 0 {
+ return 0, errors.New("encryption: ReadAt: negative offset")
+ }
+ if off >= f.plaintextLength {
+ return 0, io.EOF
+ }
+
+ var read int
+ for read < len(p) {
+ curOff := off + int64(read)
+ if curOff >= f.plaintextLength {
+ break
+ }
+ idx := curOff / gcmStreamPlainBlockSize
+ block, err := f.readBlock(idx)
+ if err != nil {
+ return read, err
+ }
+ inBlockOff := curOff - idx*gcmStreamPlainBlockSize
+ read += copy(p[read:], block[inBlockOff:])
+ }
+
+ var err error
+ if read < len(p) {
+ err = io.EOF
+ }
+
+ return read, err
+}
+
+func (f *standardInputFile) Read(p []byte) (int, error) {
+ n, err := f.ReadAt(p, f.pos)
+ f.pos += int64(n)
+
+ return n, err
+}
+
+func (f *standardInputFile) Seek(offset int64, whence int) (int64, error) {
+ var newPos int64
+ switch whence {
+ case io.SeekStart:
+ newPos = offset
+ case io.SeekCurrent:
+ newPos = f.pos + offset
+ case io.SeekEnd:
+ newPos = f.plaintextLength + offset
+ default:
+ return 0, fmt.Errorf("encryption: Seek: invalid whence %d", whence)
+ }
+ if newPos < 0 {
+ return 0, errors.New("encryption: Seek: negative position")
+ }
+ f.pos = newPos
+
+ return newPos, nil
+}
+
+func (f *standardInputFile) Close() error { return f.underlying.Close() }
+
+func (f *standardInputFile) Stat() (fs.FileInfo, error) {
+ info, err := f.underlying.Stat()
+ if err != nil {
+ return nil, err
+ }
+
+ return standardFileInfo{FileInfo: info, size: f.plaintextLength}, nil
+}
+
+// KeyMetadata returns the key metadata this file was decrypted with.
+func (f *standardInputFile) KeyMetadata() EncryptionKeyMetadata { return f.keyMetadata }
+
+// standardFileInfo overrides Size() to report the plaintext length rather
+// than the (larger) on-disk ciphertext length.
+type standardFileInfo struct {
+ fs.FileInfo
+ size int64
+}
+
+func (i standardFileInfo) Size() int64 { return i.size }
diff --git a/encryption/standard_manager_test.go b/encryption/standard_manager_test.go
new file mode 100644
index 000000000..49942809a
--- /dev/null
+++ b/encryption/standard_manager_test.go
@@ -0,0 +1,947 @@
+// 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_test
+
+import (
+ "bytes"
+ "crypto/aes"
+ "crypto/cipher"
+ "encoding/binary"
+ "encoding/hex"
+ "errors"
+ "fmt"
+ "io"
+ "io/fs"
+ "math"
+ mrand "math/rand/v2"
+ "slices"
+ "testing"
+ "time"
+
+ "github.com/apache/iceberg-go/encryption"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ "golang.org/x/sync/errgroup"
+)
+
+// AES GCM Stream layout constants, restated here from the spec rather than
+// imported, so the tests do not share them with the code under test.
+const (
+ plainBlockSize = 1 << 20 // the one block size Java and iceberg-rust accept
+ blockOverhead = 12 + 16 // nonce + GCM tag
+ headerLength = 4 + 4 // "AGS1" magic + little-endian block length
+ testKeyID = "table-key"
+)
+
+func newTestManager(opts ...encryption.StandardManagerOption) *encryption.StandardEncryptionManager {
+ return encryption.NewStandardEncryptionManager(opts...)
+}
+
+// testData returns n deterministic pseudo-random bytes, so that blocks differ
+// from one another and a swapped or misplaced block is detectable.
+func testData(n int) []byte {
+ r := mrand.New(mrand.NewPCG(1, 2))
+ b := make([]byte, n)
+ for i := range b {
+ b[i] = byte(r.Uint32())
+ }
+
+ return b
+}
+
+func encryptAll(t *testing.T, mgr *encryption.StandardEncryptionManager, plaintext []byte) (ciphertext []byte, keyMetadata encryption.EncryptionKeyMetadata) {
+ t.Helper()
+ fw := &memFileWriter{}
+ out, err := mgr.NewEncryptedOutputFile(t.Context(), fw, testKeyID)
+ require.NoError(t, err)
+ _, err = out.Write(plaintext)
+ require.NoError(t, err)
+ require.NoError(t, out.Close())
+
+ return fw.Bytes(), out.KeyMetadata()
+}
+
+func decryptAll(t *testing.T, mgr *encryption.StandardEncryptionManager, ciphertext []byte, keyMetadata encryption.EncryptionKeyMetadata) []byte {
+ t.Helper()
+ in, err := mgr.NewDecryptedInputFile(t.Context(), newMemFile(ciphertext), keyMetadata)
+ require.NoError(t, err)
+ data, err := io.ReadAll(in)
+ require.NoError(t, err)
+
+ return data
+}
+
+// expectedStreamLength is the length of the encrypted stream for n plaintext
+// bytes: the header, then ceil(n/block) blocks of nonce, ciphertext and tag,
+// and always at least block 0 (an empty file is a header plus one empty block).
+func expectedStreamLength(n int) int {
+ blocks := max(1, (n+plainBlockSize-1)/plainBlockSize)
+
+ return headerLength + n + blocks*blockOverhead
+}
+
+// ---------------------------------------------------------------------------
+// Independent AGS1 / StandardKeyMetadata codecs, built only from the standard
+// library and the spec text so that they share no code, and no bugs, with the
+// package under test.
+// ---------------------------------------------------------------------------
+
+func avroLong(v int64) []byte { return binary.AppendVarint(nil, v) }
+
+func avroBytes(b []byte) []byte { return append(avroLong(int64(len(b))), b...) }
+
+// avroKeyMetadata hand-encodes StandardKeyMetadata with both optional fields
+// present: version byte 1, then the Avro record
+// {encryption_key: bytes, aad_prefix: ["null", bytes], file_length: ["null", long]}.
+func avroKeyMetadata(dek, aadPrefix []byte, fileLength int64) []byte {
+ return slices.Concat([]byte{1}, avroBytes(dek), avroLong(1), avroBytes(aadPrefix), avroLong(1), avroLong(fileLength))
+}
+
+// javaKeyMetadata hand-encodes StandardKeyMetadata the way Java's
+// StandardEncryptionManager writes it for data files and manifests,
+// new StandardKeyMetadata(fileDek, aadPrefix): file_length takes the null
+// union branch.
+func javaKeyMetadata(dek, aadPrefix []byte) []byte {
+ return slices.Concat([]byte{1}, avroBytes(dek), avroLong(1), avroBytes(aadPrefix), avroLong(0))
+}
+
+type parsedKeyMetadata struct {
+ dek, aadPrefix []byte
+ fileLength int64
+}
+
+func parseKeyMetadata(t *testing.T, km []byte) parsedKeyMetadata {
+ t.Helper()
+ require.NotEmpty(t, km)
+ require.Equal(t, byte(1), km[0], "version byte")
+ rest := km[1:]
+
+ readLong := func() int64 {
+ v, n := binary.Varint(rest)
+ require.Positive(t, n, "malformed varint")
+ rest = rest[n:]
+
+ return v
+ }
+ readBytes := func() []byte {
+ n := readLong()
+ require.GreaterOrEqual(t, n, int64(0))
+ require.LessOrEqual(t, n, int64(len(rest)))
+ b := rest[:n]
+ rest = rest[n:]
+
+ return b
+ }
+
+ var p parsedKeyMetadata
+ p.dek = readBytes()
+ require.EqualValues(t, 1, readLong(), "aad_prefix must be present (union branch 1)")
+ p.aadPrefix = readBytes()
+ require.EqualValues(t, 1, readLong(), "file_length must be present (union branch 1)")
+ p.fileLength = readLong()
+ require.Empty(t, rest, "trailing bytes after the record")
+
+ return p
+}
+
+func newGCM(t *testing.T, dek []byte) cipher.AEAD {
+ t.Helper()
+ block, err := aes.NewCipher(dek)
+ require.NoError(t, err)
+ gcm, err := cipher.NewGCM(block)
+ require.NoError(t, err)
+
+ return gcm
+}
+
+// blockAAD builds the spec's AAD, aadPrefix || little-endian uint32 block index.
+func blockAAD(aadPrefix []byte, index uint32) []byte {
+ return binary.LittleEndian.AppendUint32(slices.Clone(aadPrefix), index)
+}
+
+// decodeAGS1 reads an AGS1 stream the way the spec describes.
+func decodeAGS1(t *testing.T, stream, dek, aadPrefix []byte) []byte {
+ t.Helper()
+ require.GreaterOrEqual(t, len(stream), headerLength)
+ require.Equal(t, "AGS1", string(stream[:4]), "magic")
+ require.EqualValues(t, plainBlockSize, binary.LittleEndian.Uint32(stream[4:8]), "plain block length")
+
+ gcm := newGCM(t, dek)
+ var plaintext []byte
+ body := stream[headerLength:]
+ for index := uint32(0); len(body) > 0; index++ {
+ block := body[:min(len(body), plainBlockSize+blockOverhead)]
+ body = body[len(block):]
+ require.GreaterOrEqual(t, len(block), blockOverhead, "block %d", index)
+
+ nonce, sealed := block[:12], block[12:]
+ opened, err := gcm.Open(nil, nonce, sealed, blockAAD(aadPrefix, index))
+ require.NoError(t, err, "block %d", index)
+ plaintext = append(plaintext, opened...)
+ }
+
+ return plaintext
+}
+
+// encodeAGS1 writes an AGS1 stream the way Java does: block 0 is always
+// written, even for empty plaintext. The real writer draws a random nonce per
+// block; this one derives each block's nonce from its index (so it still
+// differs from block to block) to make the output reproducible. It is only
+// fit for tests.
+func encodeAGS1(t *testing.T, plaintext, dek, aadPrefix []byte) []byte {
+ t.Helper()
+ gcm := newGCM(t, dek)
+
+ stream := append([]byte("AGS1"), binary.LittleEndian.AppendUint32(nil, plainBlockSize)...)
+ for index := uint32(0); index == 0 || len(plaintext) > 0; index++ {
+ n := min(len(plaintext), plainBlockSize)
+ nonce := bytes.Repeat([]byte{byte(index) + 1}, 12)
+ stream = gcm.Seal(append(stream, nonce...), nonce, plaintext[:n], blockAAD(aadPrefix, index))
+ plaintext = plaintext[n:]
+ }
+
+ return stream
+}
+
+// ---------------------------------------------------------------------------
+// Round trips and layout
+// ---------------------------------------------------------------------------
+
+func TestStandardEncryptionManager_RoundTrip(t *testing.T) {
+ sizes := []int{
+ 0,
+ 1,
+ 13,
+ plainBlockSize - 1,
+ plainBlockSize,
+ plainBlockSize + 1,
+ 2 * plainBlockSize,
+ 2*plainBlockSize + 12345,
+ }
+ for _, size := range sizes {
+ t.Run(fmt.Sprintf("%d bytes", size), func(t *testing.T) {
+ mgr := newTestManager()
+ plaintext := testData(size)
+
+ ciphertext, keyMetadata := encryptAll(t, mgr, plaintext)
+ assert.Len(t, ciphertext, expectedStreamLength(size), "no empty trailing block, and block 0 even when empty")
+ assert.EqualValues(t, len(ciphertext), parseKeyMetadata(t, keyMetadata).fileLength, "file_length is the encrypted length")
+
+ got := decryptAll(t, mgr, ciphertext, keyMetadata)
+ assert.True(t, bytes.Equal(plaintext, got), "decrypted bytes differ from the plaintext")
+ })
+ }
+}
+
+// Java writes block 0 even for an empty file, and iceberg-rust rejects any
+// stream shorter than the resulting 8 + 28 = 36 bytes.
+func TestStandardEncryptionManager_EmptyFileUsesJavaLayout(t *testing.T) {
+ mgr := newTestManager()
+ ciphertext, keyMetadata := encryptAll(t, mgr, nil)
+
+ assert.Len(t, ciphertext, 36)
+ assert.EqualValues(t, 36, parseKeyMetadata(t, keyMetadata).fileLength)
+
+ in, err := mgr.NewDecryptedInputFile(t.Context(), newMemFile(ciphertext), keyMetadata)
+ require.NoError(t, err)
+ info, err := in.Stat()
+ require.NoError(t, err)
+ assert.EqualValues(t, 0, info.Size())
+ got, err := io.ReadAll(in)
+ require.NoError(t, err)
+ assert.Empty(t, got)
+}
+
+func TestStandardEncryptionManager_RandomAccess(t *testing.T) {
+ mgr := newTestManager()
+ plaintext := testData(2*plainBlockSize + 100)
+
+ ciphertext, keyMetadata := encryptAll(t, mgr, plaintext)
+ in, err := mgr.NewDecryptedInputFile(t.Context(), newMemFile(ciphertext), keyMetadata)
+ require.NoError(t, err)
+
+ // Read a slice spanning a block boundary.
+ buf := make([]byte, 20)
+ n, err := in.ReadAt(buf, plainBlockSize-10)
+ require.NoError(t, err)
+ assert.Equal(t, plaintext[plainBlockSize-10:plainBlockSize+10], buf[:n])
+
+ // A read running off the end returns what is left, with io.EOF.
+ tail := make([]byte, 200)
+ n, err = in.ReadAt(tail, 2*plainBlockSize)
+ assert.Equal(t, io.EOF, err)
+ assert.Equal(t, plaintext[2*plainBlockSize:], tail[:n])
+
+ // Seek + Read.
+ pos, err := in.Seek(5, io.SeekStart)
+ require.NoError(t, err)
+ assert.Equal(t, int64(5), pos)
+
+ rest, err := io.ReadAll(in)
+ require.NoError(t, err)
+ assert.True(t, bytes.Equal(plaintext[5:], rest))
+}
+
+func TestStandardEncryptionManager_EmptyFile_ZeroLengthReadAt(t *testing.T) {
+ mgr := newTestManager()
+ ciphertext, keyMetadata := encryptAll(t, mgr, nil)
+
+ in, err := mgr.NewDecryptedInputFile(t.Context(), newMemFile(ciphertext), keyMetadata)
+ require.NoError(t, err)
+
+ n, err := in.ReadAt(nil, 0)
+ assert.NoError(t, err, "a zero-length ReadAt must return (0, nil), like os.File")
+ assert.Equal(t, 0, n)
+}
+
+func TestStandardEncryptionManager_ReadFrom(t *testing.T) {
+ mgr := newTestManager()
+ fw := &memFileWriter{}
+ out, err := mgr.NewEncryptedOutputFile(t.Context(), fw, testKeyID)
+ require.NoError(t, err)
+
+ plaintext := testData(2*plainBlockSize + 7)
+ n, err := out.ReadFrom(bytes.NewReader(plaintext))
+ require.NoError(t, err)
+ assert.EqualValues(t, len(plaintext), n)
+ require.NoError(t, out.Close())
+
+ got := decryptAll(t, mgr, fw.Bytes(), out.KeyMetadata())
+ assert.True(t, bytes.Equal(plaintext, got))
+}
+
+func TestStandardEncryptionManager_Stat(t *testing.T) {
+ mgr := newTestManager()
+ plaintext := []byte("hello iceberg world, this is a stat test")
+ ciphertext, keyMetadata := encryptAll(t, mgr, plaintext)
+
+ in, err := mgr.NewDecryptedInputFile(t.Context(), newMemFile(ciphertext), keyMetadata)
+ require.NoError(t, err)
+
+ info, err := in.Stat()
+ require.NoError(t, err)
+ assert.EqualValues(t, len(plaintext), info.Size(), "Stat must report the plaintext length, not the on-disk ciphertext length")
+ assert.Less(t, info.Size(), int64(len(ciphertext)), "ciphertext must be larger than plaintext due to the header and per-block nonce/tag overhead")
+}
+
+// ---------------------------------------------------------------------------
+// Conformance with the AGS1 spec and StandardKeyMetadata
+// ---------------------------------------------------------------------------
+
+// Round trips through the package can't catch a wrong AAD encoding (say, a
+// big-endian block index), because writer and reader share gcmStreamBlockAAD.
+// This decodes with crypto/aes and crypto/cipher alone, so such an encoding
+// fails at block 1.
+func TestStandardEncryptionManager_OutputDecodesWithSpecDecoder(t *testing.T) {
+ for _, size := range []int{0, 1, plainBlockSize, 2*plainBlockSize + 17} {
+ t.Run(fmt.Sprintf("%d bytes", size), func(t *testing.T) {
+ plaintext := testData(size)
+ ciphertext, keyMetadata := encryptAll(t, newTestManager(), plaintext)
+
+ meta := parseKeyMetadata(t, keyMetadata)
+ assert.Len(t, meta.dek, 16, "the default data key length is 128-bit")
+ assert.Len(t, meta.aadPrefix, 16)
+ assert.EqualValues(t, len(ciphertext), meta.fileLength)
+
+ got := decodeAGS1(t, ciphertext, meta.dek, meta.aadPrefix)
+ assert.True(t, bytes.Equal(plaintext, got))
+ })
+ }
+}
+
+// The reverse direction: a stream and key metadata produced by the
+// independent codecs above must be readable by the manager.
+func TestStandardEncryptionManager_ReadsSpecBuiltStream(t *testing.T) {
+ dek, err := hex.DecodeString("000102030405060708090a0b0c0d0e0f")
+ require.NoError(t, err)
+ aadPrefix, err := hex.DecodeString("a0a1a2a3a4a5a6a7a8a9aaabacadaeaf")
+ require.NoError(t, err)
+
+ // Known-answer vector: StandardKeyMetadata for an empty file, written out
+ // by hand from the Avro binary encoding. 0x01 is the version, 0x20 the
+ // zig-zag length 16, each 0x02 selects the non-null union branch, and 0x48
+ // is the zig-zag encoding of the 36-byte file length.
+ const wantEmpty = "01" + "20" + "000102030405060708090a0b0c0d0e0f" + "02" + "20" + "a0a1a2a3a4a5a6a7a8a9aaabacadaeaf" + "02" + "48"
+ assert.Equal(t, wantEmpty, hex.EncodeToString(avroKeyMetadata(dek, aadPrefix, 36)), "the test encoder must match the hand-written vector")
+
+ for _, size := range []int{0, 1, plainBlockSize, 2*plainBlockSize + 17} {
+ t.Run(fmt.Sprintf("%d bytes", size), func(t *testing.T) {
+ plaintext := testData(size)
+ stream := encodeAGS1(t, plaintext, dek, aadPrefix)
+ require.Len(t, stream, expectedStreamLength(size))
+
+ got := decryptAll(t, newTestManager(), stream, avroKeyMetadata(dek, aadPrefix, int64(len(stream))))
+ assert.True(t, bytes.Equal(plaintext, got))
+ })
+ }
+}
+
+// Java records no file_length for data files and manifests; the reader then
+// takes the encrypted length from the caller (DataFile.FileSizeBytes or
+// ManifestFile.Length).
+func TestStandardEncryptionManager_ReadsJavaKeyMetadataWithCallerLength(t *testing.T) {
+ dek, err := hex.DecodeString("000102030405060708090a0b0c0d0e0f")
+ require.NoError(t, err)
+ aadPrefix, err := hex.DecodeString("a0a1a2a3a4a5a6a7a8a9aaabacadaeaf")
+ require.NoError(t, err)
+
+ // Known-answer vector for new StandardKeyMetadata(dek, aadPrefix): the
+ // trailing 0x00 selects the null branch of the file_length union.
+ const wantJava = "01" + "20" + "000102030405060708090a0b0c0d0e0f" + "02" + "20" + "a0a1a2a3a4a5a6a7a8a9aaabacadaeaf" + "00"
+ keyMetadata := javaKeyMetadata(dek, aadPrefix)
+ assert.Equal(t, wantJava, hex.EncodeToString(keyMetadata), "the test encoder must match the hand-written vector")
+
+ mgr := newTestManager()
+ for _, size := range []int{0, 1, plainBlockSize, 2*plainBlockSize + 17} {
+ t.Run(fmt.Sprintf("%d bytes", size), func(t *testing.T) {
+ plaintext := testData(size)
+ stream := encodeAGS1(t, plaintext, dek, aadPrefix)
+
+ in, err := mgr.NewDecryptedInputFileWithLength(t.Context(), newMemFile(stream), int64(len(stream)), keyMetadata)
+ require.NoError(t, err)
+ got, err := io.ReadAll(in)
+ require.NoError(t, err)
+ assert.True(t, bytes.Equal(plaintext, got))
+
+ _, err = mgr.NewDecryptedInputFile(t.Context(), newMemFile(stream), keyMetadata)
+ assert.ErrorIs(t, err, encryption.ErrInvalidKeyMetadata, "without file_length or a caller length there is no trusted length")
+ })
+ }
+
+ t.Run("file_length takes precedence over the caller length", func(t *testing.T) {
+ plaintext := testData(plainBlockSize + 5)
+ stream := encodeAGS1(t, plaintext, dek, aadPrefix)
+ meta := avroKeyMetadata(dek, aadPrefix, int64(len(stream)))
+
+ in, err := mgr.NewDecryptedInputFileWithLength(t.Context(), newMemFile(stream), headerLength+blockOverhead, meta)
+ require.NoError(t, err)
+ got, err := io.ReadAll(in)
+ require.NoError(t, err)
+ assert.True(t, bytes.Equal(plaintext, got))
+ })
+
+ t.Run("caller length is validated", func(t *testing.T) {
+ stream := encodeAGS1(t, nil, dek, aadPrefix)
+ for _, length := range []int64{-1, 0, headerLength + blockOverhead - 1} {
+ _, err := mgr.NewDecryptedInputFileWithLength(t.Context(), newMemFile(stream), length, keyMetadata)
+ assert.ErrorIs(t, err, encryption.ErrInvalidKeyMetadata, "length %d", length)
+ }
+ })
+
+ t.Run("short caller length fails authentication", func(t *testing.T) {
+ // A caller length that drops the tail of the last block makes that
+ // block fail authentication.
+ stream := encodeAGS1(t, testData(100), dek, aadPrefix)
+ in, err := mgr.NewDecryptedInputFileWithLength(t.Context(), newMemFile(stream), int64(len(stream)-10), keyMetadata)
+ require.NoError(t, err)
+ _, err = io.ReadAll(in)
+ assert.ErrorIs(t, err, encryption.ErrAuthenticationFailed)
+ })
+}
+
+// TestStandardEncryptionManager_BlockNoncesAreUnique pins the one GCM
+// property whose loss is catastrophic and silent: reusing a nonce under a
+// single key leaks the plaintext XOR of the two blocks and the GHASH
+// authentication key, yet a file sealed with a reused (or constant) nonce
+// still round-trips and decrypts correctly, so no round-trip, tamper, or
+// reorder test would ever catch a regression here. Sealing identical
+// plaintext in every block makes any nonce collision directly visible as
+// matching ciphertext bytes.
+func TestStandardEncryptionManager_BlockNoncesAreUnique(t *testing.T) {
+ const numBlocks = 4
+ // Every block carries byte-for-byte identical plaintext.
+ plaintext := bytes.Repeat([]byte("0123456789abcdef"), numBlocks*plainBlockSize/16)
+
+ ciphertext, _ := encryptAll(t, newTestManager(), plaintext)
+ require.Len(t, ciphertext, headerLength+numBlocks*(plainBlockSize+blockOverhead))
+
+ seenNonceAt := make(map[string]int, numBlocks)
+ cipherBlocks := make([][]byte, numBlocks)
+ for i := range numBlocks {
+ offset := headerLength + i*(plainBlockSize+blockOverhead)
+ block := ciphertext[offset : offset+plainBlockSize+blockOverhead]
+ nonce := string(block[:12])
+
+ if prev, ok := seenNonceAt[nonce]; ok {
+ t.Fatalf("block %d reused the nonce from block %d: GCM nonce reuse under a single key leaks the plaintext XOR and the GHASH authentication key", i, prev)
+ }
+ seenNonceAt[nonce] = i
+ cipherBlocks[i] = block
+ }
+
+ for i := 1; i < numBlocks; i++ {
+ assert.False(t, bytes.Equal(cipherBlocks[0], cipherBlocks[i]), "identical plaintext in blocks 0 and %d must not produce identical ciphertext", i)
+ }
+}
+
+func TestStandardEncryptionManager_DataKeyLength(t *testing.T) {
+ t.Run("valid lengths", func(t *testing.T) {
+ for _, length := range []int{16, 24, 32} {
+ mgr := newTestManager(encryption.WithStandardDataKeyLength(length))
+ plaintext := testData(1000)
+
+ ciphertext, keyMetadata := encryptAll(t, mgr, plaintext)
+ assert.Len(t, parseKeyMetadata(t, keyMetadata).dek, length)
+ assert.Equal(t, plaintext, decryptAll(t, mgr, ciphertext, keyMetadata))
+ }
+ })
+
+ t.Run("invalid lengths", func(t *testing.T) {
+ for _, length := range []int{-1, 0, 15, 17, 33} {
+ mgr := newTestManager(encryption.WithStandardDataKeyLength(length))
+ _, err := mgr.NewEncryptedOutputFile(t.Context(), &memFileWriter{}, testKeyID)
+ require.Error(t, err, "length %d", length)
+ assert.ErrorIs(t, err, encryption.ErrInvalidKeyLength)
+ }
+ })
+}
+
+// ---------------------------------------------------------------------------
+// Fail-closed behaviour and untrusted input
+// ---------------------------------------------------------------------------
+
+func TestStandardEncryptionManager_OutputFile_EmptyKeyIDRejected(t *testing.T) {
+ _, err := newTestManager().NewEncryptedOutputFile(t.Context(), &memFileWriter{}, "")
+ require.Error(t, err)
+ assert.ErrorIs(t, err, encryption.ErrKeyIDRequired)
+}
+
+func TestStandardEncryptionManager_InputFile_EmptyKeyMetadataRejected(t *testing.T) {
+ _, err := newTestManager().NewDecryptedInputFile(t.Context(), newMemFile(nil), nil)
+ require.Error(t, err)
+ assert.ErrorIs(t, err, encryption.ErrKeyMetadataRequired)
+}
+
+func TestStandardEncryptionManager_TamperedBlockFailsAuthentication(t *testing.T) {
+ mgr := newTestManager()
+ ciphertext, keyMetadata := encryptAll(t, mgr, testData(2*plainBlockSize))
+ // Flip a bit inside the first block's ciphertext, past the 8-byte AES
+ // GCM Stream header and 12-byte nonce; flipping the header instead
+ // would fail earlier, at stream-header validation.
+ ciphertext[headerLength+12] ^= 0xFF
+
+ in, err := mgr.NewDecryptedInputFile(t.Context(), newMemFile(ciphertext), keyMetadata)
+ require.NoError(t, err)
+
+ _, err = io.ReadAll(in)
+ require.Error(t, err)
+ assert.ErrorIs(t, err, encryption.ErrAuthenticationFailed)
+}
+
+func TestStandardEncryptionManager_ReorderedBlocksFailAuthentication(t *testing.T) {
+ mgr := newTestManager()
+ ciphertext, keyMetadata := encryptAll(t, mgr, testData(2*plainBlockSize)) // exactly two full blocks
+
+ // Swap the two blocks: with a random nonce per block, only the AAD (which
+ // binds a block to its position) can catch this.
+ const physical = plainBlockSize + blockOverhead
+ block0 := slices.Clone(ciphertext[headerLength : headerLength+physical])
+ block1 := slices.Clone(ciphertext[headerLength+physical : headerLength+2*physical])
+ copy(ciphertext[headerLength:], block1)
+ copy(ciphertext[headerLength+physical:], block0)
+
+ in, err := mgr.NewDecryptedInputFile(t.Context(), newMemFile(ciphertext), keyMetadata)
+ require.NoError(t, err)
+
+ _, err = io.ReadAll(in)
+ require.Error(t, err)
+ assert.ErrorIs(t, err, encryption.ErrAuthenticationFailed, "swapping blocks must be detected: the AAD binds each block to its position")
+}
+
+func TestStandardEncryptionManager_WrongKeyMaterialFailsAuthentication(t *testing.T) {
+ mgr := newTestManager()
+ ciphertext, keyMetadata := encryptAll(t, mgr, []byte("hello iceberg"))
+ good := parseKeyMetadata(t, keyMetadata)
+
+ tests := map[string][]byte{
+ "wrong DEK": avroKeyMetadata(bytes.Repeat([]byte{0x55}, len(good.dek)), good.aadPrefix, good.fileLength),
+ "wrong AAD prefix": avroKeyMetadata(good.dek, bytes.Repeat([]byte{0x55}, len(good.aadPrefix)), good.fileLength),
+ }
+ for name, meta := range tests {
+ t.Run(name, func(t *testing.T) {
+ in, err := mgr.NewDecryptedInputFile(t.Context(), newMemFile(ciphertext), meta)
+ require.NoError(t, err)
+
+ _, err = io.ReadAll(in)
+ require.Error(t, err)
+ assert.ErrorIs(t, err, encryption.ErrAuthenticationFailed)
+ })
+ }
+}
+
+func TestStandardEncryptionManager_InvalidStreamHeaderRejected(t *testing.T) {
+ mgr := newTestManager()
+ ciphertext, keyMetadata := encryptAll(t, mgr, []byte("hello iceberg"))
+
+ setBlockLength := func(n uint32) func([]byte) {
+ return func(b []byte) { binary.LittleEndian.PutUint32(b[4:8], n) }
+ }
+ tests := map[string]func([]byte){
+ "magic mismatch": func(b []byte) { copy(b[:4], "XXXX") },
+ "zero block length": setBlockLength(0),
+ "64 KiB block length": setBlockLength(64 * 1024),
+ "2 MiB block length": setBlockLength(2 * plainBlockSize),
+ "maximum uint32 block length": setBlockLength(math.MaxUint32),
+ }
+ for name, mutate := range tests {
+ t.Run(name, func(t *testing.T) {
+ stream := slices.Clone(ciphertext)
+ mutate(stream)
+
+ _, err := mgr.NewDecryptedInputFile(t.Context(), newMemFile(stream), keyMetadata)
+ require.Error(t, err)
+ assert.ErrorIs(t, err, encryption.ErrInvalidStreamHeader)
+ })
+ }
+
+ t.Run("truncated header", func(t *testing.T) {
+ _, err := mgr.NewDecryptedInputFile(t.Context(), newMemFile(ciphertext[:5]), keyMetadata)
+ require.Error(t, err)
+ assert.ErrorIs(t, err, encryption.ErrInvalidStreamHeader)
+ })
+}
+
+func TestStandardEncryptionManager_InvalidKeyMetadataRejectedOnRead(t *testing.T) {
+ dek := bytes.Repeat([]byte{0x11}, 16)
+ aad := bytes.Repeat([]byte{0x22}, 16)
+ valid := func(fileLength int64) []byte { return avroKeyMetadata(dek, aad, fileLength) }
+ const physical = plainBlockSize + blockOverhead
+
+ tests := []struct {
+ name string
+ meta []byte
+ wantErr error
+ wantMsg string
+ }{
+ {"unsupported version", slices.Concat([]byte{2}, valid(36)[1:]), encryption.ErrUnsupportedKeyMetadataVersion, ""},
+ {"version byte only", []byte{1}, encryption.ErrInvalidKeyMetadata, "varint"},
+ {"truncated mid-record", valid(36)[:20], encryption.ErrInvalidKeyMetadata, "out of range"},
+ {"trailing bytes", append(valid(36), 0), encryption.ErrInvalidKeyMetadata, "trailing"},
+ {"oversized varint", slices.Concat([]byte{1}, bytes.Repeat([]byte{0xff}, 11)), encryption.ErrInvalidKeyMetadata, "varint"},
+ {"negative bytes length", slices.Concat([]byte{1}, avroLong(-1)), encryption.ErrInvalidKeyMetadata, "out of range"},
+ {"bytes length beyond the record", slices.Concat([]byte{1}, avroLong(1<<40)), encryption.ErrInvalidKeyMetadata, "out of range"},
+ {"unknown union branch", slices.Concat([]byte{1}, avroBytes(dek), avroLong(2)), encryption.ErrInvalidKeyMetadata, "union branch"},
+ {"missing aad prefix", slices.Concat([]byte{1}, avroBytes(dek), avroLong(0), avroLong(1), avroLong(36)), encryption.ErrInvalidKeyMetadata, "aad-prefix must be exactly 16 bytes, got 0"},
+ {"empty aad prefix", avroKeyMetadata(dek, nil, 36), encryption.ErrInvalidKeyMetadata, "aad-prefix must be exactly 16 bytes, got 0"},
+ {"short aad prefix", avroKeyMetadata(dek, aad[:15], 36), encryption.ErrInvalidKeyMetadata, "aad-prefix must be exactly 16 bytes, got 15"},
+ {"long aad prefix", avroKeyMetadata(dek, append(slices.Clone(aad), 0), 36), encryption.ErrInvalidKeyMetadata, "aad-prefix must be exactly 16 bytes, got 17"},
+ {"missing file length and no caller length", javaKeyMetadata(dek, aad), encryption.ErrInvalidKeyMetadata, "file-length is required"},
+ {"negative file length", valid(-1), encryption.ErrInvalidKeyMetadata, "shorter than"},
+ {"file length below the 36-byte minimum", valid(35), encryption.ErrInvalidKeyMetadata, "shorter than"},
+ {"file length ends inside a nonce", valid(headerLength + physical + 5), encryption.ErrInvalidKeyMetadata, "nonce and tag"},
+ {"invalid DEK length", avroKeyMetadata(dek[:15], aad, 36), encryption.ErrInvalidKeyLength, ""},
+ {"empty DEK", avroKeyMetadata(nil, aad, 36), encryption.ErrInvalidKeyLength, ""},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ _, err := newTestManager().NewDecryptedInputFile(t.Context(), newMemFile(nil), tt.meta)
+ require.Error(t, err)
+ assert.ErrorIs(t, err, tt.wantErr)
+ if tt.wantMsg != "" {
+ assert.ErrorContains(t, err, tt.wantMsg)
+ }
+ })
+ }
+}
+
+// A file length near math.MaxInt64 is well-formed, so construction succeeds,
+// but reading must fail closed without overflowing or allocating more than
+// one block.
+func TestStandardEncryptionManager_HugeFileLengthFailsClosedOnRead(t *testing.T) {
+ mgr := newTestManager()
+ dek := bytes.Repeat([]byte{0x11}, 16)
+ aad := bytes.Repeat([]byte{0x22}, 16)
+ header := encodeAGS1(t, nil, dek, aad)[:headerLength]
+
+ in, err := mgr.NewDecryptedInputFile(t.Context(), newMemFile(header), avroKeyMetadata(dek, aad, math.MaxInt64))
+ require.NoError(t, err)
+
+ buf := make([]byte, 16)
+ _, err = in.ReadAt(buf, 0)
+ assert.ErrorIs(t, err, encryption.ErrBlockTruncated, "block 0 lies beyond the real file")
+
+ _, err = in.ReadAt(buf, (math.MaxUint32+1)*plainBlockSize)
+ assert.ErrorIs(t, err, encryption.ErrInvalidKeyMetadata, "the AAD only has 32 bits for the block index")
+}
+
+// shortReadFile is an icebergio.File stub whose ReadAt returns fewer bytes
+// than requested along with io.EOF once truncateAt is reached, mimicking a
+// real backend (S3, local fs) reading a genuinely truncated file. A plain
+// bytes.Reader always fills the requested slice, which is why the round-trip
+// tests never exercise this path.
+type shortReadFile struct {
+ *memFile
+ truncateAt int64
+}
+
+func (f *shortReadFile) ReadAt(p []byte, off int64) (int, error) {
+ if off >= f.truncateAt {
+ return 0, io.EOF
+ }
+ if off+int64(len(p)) > f.truncateAt {
+ p = p[:f.truncateAt-off]
+ }
+ n, err := f.memFile.ReadAt(p, off)
+ if err == nil && int64(n) < int64(len(p)) {
+ err = io.EOF
+ }
+
+ return n, err
+}
+
+func TestStandardEncryptionManager_TruncatedBackendReadReportsBlockTruncated(t *testing.T) {
+ mgr := newTestManager()
+ ciphertext, keyMetadata := encryptAll(t, mgr, testData(100))
+ truncated := &shortReadFile{memFile: newMemFile(ciphertext), truncateAt: int64(len(ciphertext) - 5)}
+
+ in, err := mgr.NewDecryptedInputFile(t.Context(), truncated, keyMetadata)
+ require.NoError(t, err)
+
+ _, err = io.ReadAll(in)
+ require.Error(t, err)
+ assert.ErrorIs(t, err, encryption.ErrBlockTruncated)
+ assert.NotErrorIs(t, err, encryption.ErrCiphertextTooShort, "a truncated block read is distinct from a too-short KMS-wrapped key/payload")
+ assert.NotErrorIs(t, err, encryption.ErrAuthenticationFailed, "a truncated read must not be misreported as tampering")
+}
+
+// The plaintext length is derived from the file length in key metadata, so
+// bytes appended after the last block are ignored.
+func TestStandardEncryptionManager_TrailingBytesAfterLastBlockIgnored(t *testing.T) {
+ mgr := newTestManager()
+ plaintext := testData(plainBlockSize + 10)
+ ciphertext, keyMetadata := encryptAll(t, mgr, plaintext)
+
+ got := decryptAll(t, mgr, append(ciphertext, []byte("trailing garbage")...), keyMetadata)
+ assert.True(t, bytes.Equal(plaintext, got))
+}
+
+// ---------------------------------------------------------------------------
+// Write and close failures
+// ---------------------------------------------------------------------------
+
+// failAfterNWriter is an icebergio.FileWriter stub that fails the Nth call
+// to Write, simulating a mid-stream flush failure on the underlying storage.
+// It also counts Close calls, since memFileWriter.Close is a no-op and would
+// otherwise mask a leaked/never-closed underlying writer.
+type failAfterNWriter struct {
+ memFileWriter
+ failAt int
+ writes int
+ closes int
+}
+
+func (w *failAfterNWriter) Write(p []byte) (int, error) {
+ w.writes++
+ if w.writes == w.failAt {
+ return 0, errors.New("simulated write failure")
+ }
+
+ return w.memFileWriter.Write(p)
+}
+
+func (w *failAfterNWriter) Close() error {
+ w.closes++
+
+ return w.memFileWriter.Close()
+}
+
+func TestStandardEncryptionManager_FlushFailurePoisonsWriterAndClose(t *testing.T) {
+ mgr := newTestManager()
+ // Write call 1 is the AES GCM Stream header written by
+ // NewEncryptedOutputFile; call 2 flushes the first block; call 3 flushes
+ // the second block and fails.
+ fw := &failAfterNWriter{failAt: 3}
+ out, err := mgr.NewEncryptedOutputFile(t.Context(), fw, testKeyID)
+ require.NoError(t, err)
+
+ // The first block flushes fine; the second block's flush fails.
+ n, err := out.Write(make([]byte, 2*plainBlockSize))
+ require.Error(t, err)
+ assert.Equal(t, plainBlockSize, n, "only the first block reached storage")
+ assert.Equal(t, 1, fw.closes, "a poisoned flush must close the underlying writer, not leak it")
+
+ // A subsequent Write must return the same sticky error, not attempt more
+ // I/O, and not the post-Close error: the writer was poisoned, not closed.
+ _, err2 := out.Write([]byte("c"))
+ require.Error(t, err2)
+ assert.ErrorIs(t, err2, err)
+ assert.NotErrorIs(t, err2, encryption.ErrOutputFileClosed)
+ assert.NotErrorIs(t, err2, fs.ErrClosed)
+
+ // Close must report the failure rather than silently succeeding.
+ closeErr := out.Close()
+ require.Error(t, closeErr)
+ assert.Nil(t, out.KeyMetadata(), "key metadata must not be finalized when the file failed to write")
+
+ // A retried Close must keep reporting the same error, not nil.
+ closeErr2 := out.Close()
+ require.Error(t, closeErr2)
+ assert.Equal(t, 1, fw.closes, "a retried Close must not close the underlying writer again")
+}
+
+func TestStandardEncryptionManager_HeaderWriteFailureClosesWriter(t *testing.T) {
+ fw := &failAfterNWriter{failAt: 1}
+ out, err := newTestManager().NewEncryptedOutputFile(t.Context(), fw, testKeyID)
+ require.Error(t, err)
+ assert.Nil(t, out)
+ assert.Equal(t, 1, fw.closes, "a failed header write must close the underlying writer, not leak it")
+}
+
+func TestStandardEncryptionManager_WriteAfterCloseRejected(t *testing.T) {
+ out, err := newTestManager().NewEncryptedOutputFile(t.Context(), &memFileWriter{}, testKeyID)
+ require.NoError(t, err)
+ require.NoError(t, out.Close())
+
+ _, err = out.Write([]byte("late"))
+ require.Error(t, err)
+ assert.ErrorIs(t, err, encryption.ErrOutputFileClosed)
+ assert.ErrorIs(t, err, fs.ErrClosed)
+}
+
+// closeFailWriter is an icebergio.FileWriter stub whose Close always fails
+// even though every Write succeeds, simulating a backend where the final
+// flush/commit (e.g. an S3 multipart completion) fails independently of the
+// preceding writes.
+type closeFailWriter struct {
+ memFileWriter
+ closes int
+}
+
+func (w *closeFailWriter) Close() error {
+ w.closes++
+
+ return errors.New("simulated close failure")
+}
+
+func TestStandardEncryptionManager_UnderlyingCloseFailurePoisonsWriter(t *testing.T) {
+ fw := &closeFailWriter{}
+ out, err := newTestManager().NewEncryptedOutputFile(t.Context(), fw, testKeyID)
+ require.NoError(t, err)
+
+ _, err = out.Write([]byte("hello"))
+ require.NoError(t, err)
+
+ closeErr := out.Close()
+ require.Error(t, closeErr)
+ assert.Nil(t, out.KeyMetadata(), "key metadata must not be finalized when the underlying Close fails")
+
+ // A retried Close must keep reporting the same error, not nil, and must
+ // not attempt to close the underlying writer again.
+ closeErr2 := out.Close()
+ require.Error(t, closeErr2)
+ assert.ErrorIs(t, closeErr2, closeErr)
+ assert.Equal(t, 1, fw.closes, "a retried Close must not re-close the underlying writer")
+
+ _, err = out.Write([]byte("late"))
+ require.Error(t, err)
+ assert.ErrorIs(t, err, closeErr, "a subsequent Write must report the same sticky error")
+}
+
+// ---------------------------------------------------------------------------
+// Block cache and concurrency
+// ---------------------------------------------------------------------------
+
+// countingReadAtFile wraps a memFile and counts ReadAt calls, to verify that
+// repeated small reads landing in the same block reuse the decrypted-block
+// cache instead of re-decrypting on every call.
+type countingReadAtFile struct {
+ *memFile
+ reads int
+}
+
+func (f *countingReadAtFile) ReadAt(p []byte, off int64) (int, error) {
+ f.reads++
+
+ return f.memFile.ReadAt(p, off)
+}
+
+func TestStandardEncryptionManager_RepeatedSmallReadsReuseDecryptedBlockCache(t *testing.T) {
+ mgr := newTestManager()
+ plaintext := testData(plainBlockSize + 5000) // a full block, then a 5000-byte tail block
+
+ ciphertext, keyMetadata := encryptAll(t, mgr, plaintext)
+ counting := &countingReadAtFile{memFile: newMemFile(ciphertext)}
+
+ in, err := mgr.NewDecryptedInputFile(t.Context(), counting, keyMetadata)
+ require.NoError(t, err)
+ baseline := counting.reads // the stream header was already read once above
+
+ buf := make([]byte, 1)
+ readOneByteAt := func(off int64) {
+ _, err := in.Seek(off, io.SeekStart)
+ require.NoError(t, err)
+ n, err := in.Read(buf)
+ require.NoError(t, err)
+ require.Equal(t, 1, n)
+ }
+ for off := range int64(5000) {
+ readOneByteAt(off) // all in block 0
+ }
+ for off := range int64(5000) {
+ readOneByteAt(plainBlockSize + off) // all in block 1
+ }
+
+ assert.LessOrEqual(t, counting.reads-baseline, 2, "10000 one-byte reads across two blocks must decrypt each block at most once")
+}
+
+// slowReadFile wraps a memFile and sleeps before every ReadAt, simulating a
+// remote backend (e.g. S3) with real network latency. It is used to check
+// that concurrent ReadAt calls for distinct blocks overlap instead of queuing
+// behind readBlock's cache lock.
+type slowReadFile struct {
+ *memFile
+ delay time.Duration
+}
+
+func (f *slowReadFile) ReadAt(p []byte, off int64) (int, error) {
+ time.Sleep(f.delay)
+
+ return f.memFile.ReadAt(p, off)
+}
+
+func TestStandardEncryptionManager_ConcurrentReadAtOverlaps(t *testing.T) {
+ const (
+ numBlocks = 8
+ delay = 20 * time.Millisecond
+ )
+ mgr := newTestManager()
+ ciphertext, keyMetadata := encryptAll(t, mgr, testData(numBlocks*plainBlockSize))
+
+ slow := &slowReadFile{memFile: newMemFile(ciphertext), delay: delay}
+ in, err := mgr.NewDecryptedInputFile(t.Context(), slow, keyMetadata) // reads the header; not timed below
+ require.NoError(t, err)
+
+ start := time.Now()
+ var g errgroup.Group
+ for i := range numBlocks {
+ g.Go(func() error {
+ buf := make([]byte, 16)
+ _, err := in.ReadAt(buf, int64(i)*plainBlockSize)
+
+ return err
+ })
+ }
+ require.NoError(t, g.Wait())
+ elapsed := time.Since(start)
+
+ // Serial execution (for example, holding cacheMu across ReadAt) would take
+ // roughly numBlocks*delay. True concurrency keeps it close to a single
+ // delay; the threshold below is generous to stay robust on slow or loaded
+ // CI machines while still failing on a serialized implementation.
+ assert.Less(t, elapsed, numBlocks/2*delay, "concurrent ReadAt calls for distinct blocks must overlap, not serialize")
+}