Skip to content

key_metadata

Key metadata for encrypted manifests, manifest lists and data files.

StandardKeyMetadata dataclass

The key and AAD prefix needed to decrypt a single file.

Wire format is a version byte followed by an Avro datum of KEY_METADATA_SCHEMA_V1, byte-compatible with Java's StandardKeyMetadata.

Source code in pyiceberg/encryption/key_metadata.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@dataclass(frozen=True)
class StandardKeyMetadata:
    """The key and AAD prefix needed to decrypt a single file.

    Wire format is a version byte followed by an Avro datum of `KEY_METADATA_SCHEMA_V1`,
    byte-compatible with Java's `StandardKeyMetadata`.
    """

    encryption_key: bytes = field(repr=False)
    aad_prefix: bytes | None = None
    file_length: int | None = None

    def __post_init__(self) -> None:
        """Reject invalid key lengths here rather than only on decode, so an invalid instance cannot exist."""
        if len(self.encryption_key) not in AES_KEY_LENGTHS:
            raise ValueError(
                f"Invalid encryption key in key metadata: expected one of {AES_KEY_LENGTHS} bytes, got {len(self.encryption_key)}"
            )

    @classmethod
    def from_bytes(cls, data: bytes) -> StandardKeyMetadata:
        """Decode key metadata from its wire format."""
        if not data:
            raise ValueError("Empty key metadata")

        if (version := data[0]) != KEY_METADATA_V1:
            raise ValueError(f"Unsupported key metadata version: {version}")

        record = construct_reader(KEY_METADATA_SCHEMA_V1).read(new_decoder(data[1:]))
        return cls(encryption_key=record[0], aad_prefix=record[1], file_length=record[2])

    def to_bytes(self) -> bytes:
        """Encode key metadata to its wire format."""
        output = io.BytesIO()
        encoder = BinaryEncoder(output)
        encoder.write(bytes([KEY_METADATA_V1]))
        record = Record(self.encryption_key, self.aad_prefix, self.file_length)
        construct_writer(KEY_METADATA_SCHEMA_V1).write(encoder, record)
        return output.getvalue()

__post_init__()

Reject invalid key lengths here rather than only on decode, so an invalid instance cannot exist.

Source code in pyiceberg/encryption/key_metadata.py 
54
55
56
57
58
59
def __post_init__(self) -> None:
    """Reject invalid key lengths here rather than only on decode, so an invalid instance cannot exist."""
    if len(self.encryption_key) not in AES_KEY_LENGTHS:
        raise ValueError(
            f"Invalid encryption key in key metadata: expected one of {AES_KEY_LENGTHS} bytes, got {len(self.encryption_key)}"
        )

from_bytes(data) classmethod

Decode key metadata from its wire format.

Source code in pyiceberg/encryption/key_metadata.py 
61
62
63
64
65
66
67
68
69
70
71
@classmethod
def from_bytes(cls, data: bytes) -> StandardKeyMetadata:
    """Decode key metadata from its wire format."""
    if not data:
        raise ValueError("Empty key metadata")

    if (version := data[0]) != KEY_METADATA_V1:
        raise ValueError(f"Unsupported key metadata version: {version}")

    record = construct_reader(KEY_METADATA_SCHEMA_V1).read(new_decoder(data[1:]))
    return cls(encryption_key=record[0], aad_prefix=record[1], file_length=record[2])

to_bytes()

Encode key metadata to its wire format.

Source code in pyiceberg/encryption/key_metadata.py 
73
74
75
76
77
78
79
80
def to_bytes(self) -> bytes:
    """Encode key metadata to its wire format."""
    output = io.BytesIO()
    encoder = BinaryEncoder(output)
    encoder.write(bytes([KEY_METADATA_V1]))
    record = Record(self.encryption_key, self.aad_prefix, self.file_length)
    construct_writer(KEY_METADATA_SCHEMA_V1).write(encoder, record)
    return output.getvalue()