_parquet_encryption.pyx

# 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.

# cython: profile=False
# distutils: language = c++

from datetime import timedelta

from cpython.bytes cimport PyBytes_FromStringAndSize
from cython.operator cimport dereference as deref

from pyarrow.includes.common cimport *
from pyarrow.includes.libarrow cimport *
from pyarrow.lib cimport check_status
from pyarrow.lib cimport _Weakrefable
from pyarrow.lib import tobytes, frombytes
from pyarrow._fs cimport FileSystem
from pyarrow.fs import _resolve_filesystem_and_path

cdef ParquetCipher cipher_from_name(name):
    name = name.upper()
    if name == 'AES_GCM_V1':
        return ParquetCipher_AES_GCM_V1
    elif name == 'AES_GCM_CTR_V1':
        return ParquetCipher_AES_GCM_CTR_V1
    else:
        raise ValueError(f'Invalid cipher name: {name!r}')


cdef cipher_to_name(ParquetCipher cipher):
    if ParquetCipher_AES_GCM_V1 == cipher:
        return 'AES_GCM_V1'
    elif ParquetCipher_AES_GCM_CTR_V1 == cipher:
        return 'AES_GCM_CTR_V1'
    else:
        raise ValueError(f'Invalid cipher value: {cipher}')

cdef class EncryptionConfiguration(_Weakrefable):
    """Configuration of the encryption, such as which columns to encrypt"""
    # Avoid mistakingly creating attributes
    __slots__ = ()

    def __init__(self, footer_key, *, column_keys=None,
                 uniform_encryption=None,
                 encryption_algorithm=None,
                 plaintext_footer=None, double_wrapping=None,
                 cache_lifetime=None, internal_key_material=None,
                 data_key_length_bits=None):
        self.configuration.reset(
            new CEncryptionConfiguration(tobytes(footer_key)))
        if column_keys is not None:
            self.column_keys = column_keys
        if uniform_encryption is not None:
            self.uniform_encryption = uniform_encryption
        if encryption_algorithm is not None:
            self.encryption_algorithm = encryption_algorithm
        if plaintext_footer is not None:
            self.plaintext_footer = plaintext_footer
        if double_wrapping is not None:
            self.double_wrapping = double_wrapping
        if cache_lifetime is not None:
            self.cache_lifetime = cache_lifetime
        if internal_key_material is not None:
            self.internal_key_material = internal_key_material
        if data_key_length_bits is not None:
            self.data_key_length_bits = data_key_length_bits

    @property
    def footer_key(self):
        """ID of the master key for footer encryption/signing"""
        return frombytes(self.configuration.get().footer_key)

    @property
    def column_keys(self):
        """
        List of columns to encrypt, with master key IDs.
        """
        column_keys_str = frombytes(self.configuration.get().column_keys)
        # Convert from "masterKeyID:colName,colName;masterKeyID:colName..."
        # (see HIVE-21848) to dictionary of master key ID to column name lists
        column_keys_to_key_list_str = dict(subString.replace(" ", "").split(
            ":") for subString in column_keys_str.split(";"))
        column_keys_dict = {k: v.split(
            ",") for k, v in column_keys_to_key_list_str.items()}
        return column_keys_dict

    @column_keys.setter
    def column_keys(self, dict value):
        if value is not None:
            # convert a dictionary such as
            # '{"key1": ["col1 ", "col2"], "key2": ["col3 ", "col4"]}''
            # to the string defined by the spec
            # 'key1: col1 , col2; key2: col3 , col4'
            column_keys = "; ".join(
                [f"{k}: {', '.join(v)}" for k, v in value.items()])
            self.configuration.get().column_keys = tobytes(column_keys)

    @property
    def uniform_encryption(self):
        """Whether to encrypt footer and all columns with the same encryption key.

        This cannot be used together with column_keys.
        """
        return self.configuration.get().uniform_encryption

    @uniform_encryption.setter
    def uniform_encryption(self, value):
        self.configuration.get().uniform_encryption = value

    @property
    def encryption_algorithm(self):
        """Parquet encryption algorithm.
        Can be "AES_GCM_V1" (default), or "AES_GCM_CTR_V1"."""
        return cipher_to_name(self.configuration.get().encryption_algorithm)

    @encryption_algorithm.setter
    def encryption_algorithm(self, value):
        cipher = cipher_from_name(value)
        self.configuration.get().encryption_algorithm = cipher

    @property
    def plaintext_footer(self):
        """Write files with plaintext footer."""
        return self.configuration.get().plaintext_footer

    @plaintext_footer.setter
    def plaintext_footer(self, value):
        self.configuration.get().plaintext_footer = value

    @property
    def double_wrapping(self):
        """Use double wrapping - where data encryption keys (DEKs) are
        encrypted with key encryption keys (KEKs), which in turn are
        encrypted with master keys.
        If set to false, use single wrapping - where DEKs are
        encrypted directly with master keys."""
        return self.configuration.get().double_wrapping

    @double_wrapping.setter
    def double_wrapping(self, value):
        self.configuration.get().double_wrapping = value

    @property
    def cache_lifetime(self):
        """Lifetime of cached entities (key encryption keys,
        local wrapping keys, KMS client objects)."""
        return timedelta(
            seconds=self.configuration.get().cache_lifetime_seconds)

    @cache_lifetime.setter
    def cache_lifetime(self, value):
        if not isinstance(value, timedelta):
            raise TypeError("cache_lifetime should be a timedelta")
        self.configuration.get().cache_lifetime_seconds = value.total_seconds()

    @property
    def internal_key_material(self):
        """Store key material inside Parquet file footers; this mode doesn’t
        produce additional files. If set to false, key material is stored in
        separate files in the same folder, which enables key rotation for
        immutable Parquet files."""
        return self.configuration.get().internal_key_material

    @internal_key_material.setter
    def internal_key_material(self, value):
        self.configuration.get().internal_key_material = value

    @property
    def data_key_length_bits(self):
        """Length of data encryption keys (DEKs), randomly generated by parquet key
        management tools. Can be 128, 192 or 256 bits."""
        return self.configuration.get().data_key_length_bits

    @data_key_length_bits.setter
    def data_key_length_bits(self, value):
        self.configuration.get().data_key_length_bits = value

    cdef inline shared_ptr[CEncryptionConfiguration] unwrap(self) nogil:
        return self.configuration


cdef class DecryptionConfiguration(_Weakrefable):
    """Configuration of the decryption, such as cache timeout."""
    # Avoid mistakingly creating attributes
    __slots__ = ()

    def __init__(self, *, cache_lifetime=None):
        self.configuration.reset(new CDecryptionConfiguration())

    @property
    def cache_lifetime(self):
        """Lifetime of cached entities (key encryption keys,
        local wrapping keys, KMS client objects)."""
        return timedelta(
            seconds=self.configuration.get().cache_lifetime_seconds)

    @cache_lifetime.setter
    def cache_lifetime(self, value):
        self.configuration.get().cache_lifetime_seconds = value.total_seconds()

    cdef inline shared_ptr[CDecryptionConfiguration] unwrap(self) nogil:
        return self.configuration


cdef class KmsConnectionConfig(_Weakrefable):
    """Configuration of the connection to the Key Management Service (KMS)"""
    # Avoid mistakingly creating attributes
    __slots__ = ()

    def __init__(self, *, kms_instance_id=None, kms_instance_url=None,
                 key_access_token=None, custom_kms_conf=None):
        self.configuration.reset(new CKmsConnectionConfig())
        if kms_instance_id is not None:
            self.kms_instance_id = kms_instance_id
        if kms_instance_url is not None:
            self.kms_instance_url = kms_instance_url
        if key_access_token is None:
            self.key_access_token = b'DEFAULT'
        else:
            self.key_access_token = key_access_token
        if custom_kms_conf is not None:
            self.custom_kms_conf = custom_kms_conf

    @property
    def kms_instance_id(self):
        """ID of the KMS instance that will be used for encryption
        (if multiple KMS instances are available)."""
        return frombytes(self.configuration.get().kms_instance_id)

    @kms_instance_id.setter
    def kms_instance_id(self, value):
        self.configuration.get().kms_instance_id = tobytes(value)

    @property
    def kms_instance_url(self):
        """URL of the KMS instance."""
        return frombytes(self.configuration.get().kms_instance_url)

    @kms_instance_url.setter
    def kms_instance_url(self, value):
        self.configuration.get().kms_instance_url = tobytes(value)

    @property
    def key_access_token(self):
        """Authorization token that will be passed to KMS."""
        return frombytes(self.configuration.get()
                         .refreshable_key_access_token.get().value())

    @key_access_token.setter
    def key_access_token(self, value):
        self.refresh_key_access_token(value)

    @property
    def custom_kms_conf(self):
        """A dictionary with KMS-type-specific configuration"""
        custom_kms_conf = {
            frombytes(k): frombytes(v)
            for k, v in self.configuration.get().custom_kms_conf
        }
        return custom_kms_conf

    @custom_kms_conf.setter
    def custom_kms_conf(self, dict value):
        if value is not None:
            for k, v in value.items():
                if isinstance(k, str) and isinstance(v, str):
                    self.configuration.get().custom_kms_conf[tobytes(k)] = \
                        tobytes(v)
                else:
                    raise TypeError("Expected custom_kms_conf to be " +
                                    "a dictionary of strings")

    def refresh_key_access_token(self, value):
        cdef:
            shared_ptr[CKeyAccessToken] c_key_access_token = \
                self.configuration.get().refreshable_key_access_token

        c_key_access_token.get().Refresh(tobytes(value))

    cdef inline shared_ptr[CKmsConnectionConfig] unwrap(self) nogil:
        return self.configuration

    @staticmethod
    cdef wrap(const CKmsConnectionConfig& config):
        result = KmsConnectionConfig()
        # We require a copy of the config because the input is
        # a const reference owned by C++.
        cdef CKmsConnectionConfig config_copy = config
        result.configuration = make_shared[CKmsConnectionConfig](move(config_copy))
        return result


# Callback definitions for CPyKmsClientVtable
cdef void _cb_wrap_key(
        handler, const CSecureString& key,
        const c_string& master_key_identifier, c_string* out) except *:
    view = <cpp_string_view>key.as_view()
    key_bytes = <bytes>PyBytes_FromStringAndSize(view.data(), view.size())
    mkid_str = frombytes(master_key_identifier)
    wrapped_key = handler.wrap_key(key_bytes, mkid_str)
    out[0] = tobytes(wrapped_key)


cdef void _cb_unwrap_key(
        handler, const c_string& wrapped_key,
        const c_string& master_key_identifier, CSecureString* out) except *:
    mkid_str = frombytes(master_key_identifier)
    wk_str = frombytes(wrapped_key)
    key = handler.unwrap_key(wk_str, mkid_str)
    cstr = <c_string>tobytes(key)
    out[0] = CSecureString(move(cstr))


cdef class KmsClient(_Weakrefable):
    """The abstract base class for KmsClient implementations."""
    cdef:
        shared_ptr[CKmsClient] client

    def __init__(self):
        self.init()

    cdef init(self):
        cdef:
            CPyKmsClientVtable vtable = CPyKmsClientVtable()

        vtable.wrap_key = _cb_wrap_key
        vtable.unwrap_key = _cb_unwrap_key

        self.client.reset(new CPyKmsClient(self, vtable))

    def wrap_key(self, key_bytes, master_key_identifier):
        """Wrap a key - encrypt it with the master key."""
        raise NotImplementedError()

    def unwrap_key(self, wrapped_key, master_key_identifier):
        """Unwrap a key - decrypt it with the master key."""
        raise NotImplementedError()

    cdef inline shared_ptr[CKmsClient] unwrap(self) nogil:
        return self.client


# Callback definition for CPyKmsClientFactoryVtable
cdef void _cb_create_kms_client(
        handler,
        const CKmsConnectionConfig& kms_connection_config,
        shared_ptr[CKmsClient]* out) except *:
    connection_config = KmsConnectionConfig.wrap(kms_connection_config)

    result = handler(connection_config)
    if not isinstance(result, KmsClient):
        raise TypeError(
            f"callable must return KmsClient instances, but got {type(result)}")

    out[0] = (<KmsClient> result).unwrap()


cdef inline shared_ptr[CFileSystem] _unwrap_fs(filesystem: FileSystem | None):
    if isinstance(filesystem, FileSystem):
        return filesystem.unwrap()
    else:
        return <shared_ptr[CFileSystem]>nullptr


cdef class CryptoFactory(_Weakrefable):
    """ A factory that produces the low-level FileEncryptionProperties and
    FileDecryptionProperties objects, from the high-level parameters."""
    # Avoid mistakingly creating attributes
    __slots__ = ()

    def __init__(self, kms_client_factory):
        """Create CryptoFactory.

        Parameters
        ----------
        kms_client_factory : a callable that accepts KmsConnectionConfig
            and returns a KmsClient
        """
        self.factory.reset(new CPyCryptoFactory())

        if callable(kms_client_factory):
            self.init(kms_client_factory)
        else:
            raise TypeError("Parameter kms_client_factory must be a callable")

    cdef init(self, callable_client_factory):
        cdef:
            CPyKmsClientFactoryVtable vtable
            shared_ptr[CPyKmsClientFactory] kms_client_factory

        vtable.create_kms_client = _cb_create_kms_client
        kms_client_factory.reset(
            new CPyKmsClientFactory(callable_client_factory, vtable))
        # A KmsClientFactory object must be registered
        # via this method before calling any of
        # file_encryption_properties()/file_decryption_properties() methods.
        self.factory.get().RegisterKmsClientFactory(
            static_pointer_cast[CKmsClientFactory, CPyKmsClientFactory](
                kms_client_factory))

    def file_encryption_properties(self,
                                   KmsConnectionConfig kms_connection_config,
                                   EncryptionConfiguration encryption_config,
                                   parquet_file_path=None,
                                   FileSystem filesystem=None):
        """Create file encryption properties.

        Parameters
        ----------
        kms_connection_config : KmsConnectionConfig
            Configuration of connection to KMS

        encryption_config : EncryptionConfiguration
            Configuration of the encryption, such as which columns to encrypt

        parquet_file_path : str, pathlib.Path, or None, default None
            Path to the parquet file to be encrypted. Only required when the
            internal_key_material attribute of EncryptionConfiguration is set
            to False. Used to derive the path for storing key material
            specific to this parquet file.

        filesystem : FileSystem or None, default None
            Used only when internal_key_material is set to False on
            EncryptionConfiguration. If None, the file system will be inferred
            based on parquet_file_path.

        Returns
        -------
        file_encryption_properties : FileEncryptionProperties
            File encryption properties.
        """
        cdef:
            CResult[shared_ptr[CFileEncryptionProperties]] \
                file_encryption_properties_result
            c_string c_parquet_file_path
            shared_ptr[CFileSystem] c_filesystem

        filesystem, parquet_file_path = _resolve_filesystem_and_path(
            parquet_file_path, filesystem)
        if parquet_file_path is not None:
            c_parquet_file_path = tobytes(parquet_file_path)
        else:
            c_parquet_file_path = tobytes("")
        c_filesystem = _unwrap_fs(filesystem)

        with nogil:
            file_encryption_properties_result = \
                self.factory.get().SafeGetFileEncryptionProperties(
                    deref(kms_connection_config.unwrap().get()),
                    deref(encryption_config.unwrap().get()),
                    c_parquet_file_path, c_filesystem)
        file_encryption_properties = GetResultValue(
            file_encryption_properties_result)
        return FileEncryptionProperties.wrap(file_encryption_properties)

    def file_decryption_properties(
            self,
            KmsConnectionConfig kms_connection_config,
            DecryptionConfiguration decryption_config=None,
            parquet_file_path=None,
            FileSystem filesystem=None):
        """Create file decryption properties.

        Parameters
        ----------
        kms_connection_config : KmsConnectionConfig
            Configuration of connection to KMS

        decryption_config : DecryptionConfiguration, default None
            Configuration of the decryption, such as cache timeout.
            Can be None.

        parquet_file_path : str, pathlib.Path, or None, default None
            Path to the parquet file to be decrypted. Only required when
            the parquet file uses external key material.  Used to derive
            the path to the external key material file.

        filesystem : FileSystem or None, default None
            Used only when the parquet file uses external key material. If
            None, the file system will be inferred based on parquet_file_path.

        Returns
        -------
        file_decryption_properties : FileDecryptionProperties
            File decryption properties.
        """
        cdef:
            CDecryptionConfiguration c_decryption_config
            CResult[shared_ptr[CFileDecryptionProperties]] \
                c_file_decryption_properties
            c_string c_parquet_file_path
            shared_ptr[CFileSystem] c_filesystem

        filesystem, parquet_file_path = _resolve_filesystem_and_path(
            parquet_file_path, filesystem)
        if parquet_file_path is not None:
            c_parquet_file_path = tobytes(parquet_file_path)
        else:
            c_parquet_file_path = tobytes("")
        c_filesystem = _unwrap_fs(filesystem)

        if decryption_config is None:
            c_decryption_config = CDecryptionConfiguration()
        else:
            c_decryption_config = deref(decryption_config.unwrap().get())
        with nogil:
            c_file_decryption_properties = \
                self.factory.get().SafeGetFileDecryptionProperties(
                    deref(kms_connection_config.unwrap().get()),
                    c_decryption_config, c_parquet_file_path, c_filesystem)
        file_decryption_properties = GetResultValue(
            c_file_decryption_properties)
        return FileDecryptionProperties.wrap(file_decryption_properties)

    def remove_cache_entries_for_token(self, access_token):
        self.factory.get().RemoveCacheEntriesForToken(tobytes(access_token))

    def remove_cache_entries_for_all_tokens(self):
        self.factory.get().RemoveCacheEntriesForAllTokens()

    def rotate_master_keys(
            self,
            KmsConnectionConfig kms_connection_config,
            parquet_file_path,
            FileSystem filesystem=None,
            double_wrapping=True,
            cache_lifetime_seconds=600):
        """ Rotates master encryption keys for a Parquet file that uses
        external key material.

        Parameters
        ----------
        kms_connection_config : KmsConnectionConfig
            Configuration of connection to KMS

        parquet_file_path : str or pathlib.Path
            Path to a parquet file using external key material.

        filesystem : FileSystem or None, default None
            Used only when the parquet file uses external key material. If
            None, the file system will be inferred based on parquet_file_path.

        double_wrapping : bool, default True
            In the single wrapping mode, encrypts data encryption keys with
            new master keys. In the double wrapping mode, generates new
            KEKs (key encryption keys) and uses these to encrypt the data keys,
            and encrypts the KEKs with the new master keys.

        cache_lifetime_seconds : int or float, default 600
            During key rotation, KMS Client and Key Encryption Keys will be
            cached for this duration.
        """
        cdef:
            c_string c_parquet_file_path
            shared_ptr[CFileSystem] c_filesystem

        if parquet_file_path != "":
            filesystem, parquet_file_path = _resolve_filesystem_and_path(
                parquet_file_path, filesystem)

        c_parquet_file_path = tobytes(parquet_file_path)
        c_filesystem = _unwrap_fs(filesystem)

        status = self.factory.get().SafeRotateMasterKeys(
            deref(kms_connection_config.unwrap().get()),
            c_parquet_file_path,
            c_filesystem,
            double_wrapping,
            cache_lifetime_seconds)

        check_status(status)

    cdef inline shared_ptr[CPyCryptoFactory] unwrap(self):
        return self.factory

cdef class KeyMaterial(_Weakrefable):

    @property
    def is_footer_key(self):
        return self.key_material.get().is_footer_key()

    @property
    def is_double_wrapped(self):
        return self.key_material.get().is_double_wrapped()

    @property
    def master_key_id(self):
        return frombytes(self.key_material.get().master_key_id())

    @property
    def wrapped_dek(self):
        return frombytes(self.key_material.get().wrapped_dek())

    @property
    def kek_id(self):
        return frombytes(self.key_material.get().kek_id())

    @property
    def wrapped_kek(self):
        return frombytes(self.key_material.get().wrapped_kek())

    @property
    def kms_instance_id(self):
        return frombytes(self.key_material.get().kms_instance_id())

    @property
    def kms_instance_url(self):
        return frombytes(self.key_material.get().kms_instance_url())

    @staticmethod
    cdef inline KeyMaterial wrap(shared_ptr[CKeyMaterial] key_material):
        wrapper = KeyMaterial()
        wrapper.key_material = key_material
        return wrapper

    @staticmethod
    def parse(
            const c_string key_material_string):
        cdef:
            shared_ptr[CKeyMaterial] c_key_material
        c_key_material = make_shared[CKeyMaterial](move(
            CKeyMaterial.Parse(key_material_string)
        ))
        return KeyMaterial.wrap(c_key_material)

cdef class FileSystemKeyMaterialStore(_Weakrefable):

    def get_key_material(self, key_id):
        cdef:
            c_string c_key_id = tobytes(key_id)
            c_string c_key_material_string

        c_key_material_string = self.store.get().GetKeyMaterial(c_key_id)
        if c_key_material_string.empty():
            raise KeyError("Invalid key id")
        return KeyMaterial.parse(c_key_material_string)

    def get_key_id_set(self):
        return self.store.get().GetKeyIDSet()

    @classmethod
    def for_file(cls, parquet_file_path,
                 FileSystem filesystem=None):
        """Creates a FileSystemKeyMaterialStore for a parquet file that
        was created with external key material.

        Parameters
        ----------
        parquet_file_path : str or pathlib.Path
            Path to a parquet file using external key material.

        filesystem : FileSystem, default None
            FileSystem where the parquet file is located. If None,
            will be inferred based on parquet_file_path.

        Returns
        -------
        FileSystemKeyMaterialStore
            A FileSystemKeyMaterialStore wrapping the external key material.
        """
        cdef:
            c_string c_parquet_file_path
            shared_ptr[CFileSystem] c_filesystem
            shared_ptr[CFileSystemKeyMaterialStore] c_store
            FileSystemKeyMaterialStore store = cls()

        filesystem, parquet_file_path = _resolve_filesystem_and_path(
            parquet_file_path, filesystem)
        c_parquet_file_path = tobytes(parquet_file_path)
        c_filesystem = _unwrap_fs(filesystem)

        c_store = CFileSystemKeyMaterialStore.Make(
            c_parquet_file_path, c_filesystem, False)
        store.store = c_store
        return store

cdef shared_ptr[CCryptoFactory] pyarrow_unwrap_cryptofactory(object crypto_factory) except *:
    if isinstance(crypto_factory, CryptoFactory):
        pycf = (<CryptoFactory> crypto_factory).unwrap()
        return static_pointer_cast[CCryptoFactory, CPyCryptoFactory](pycf)
    raise TypeError("Expected CryptoFactory, got %s" % type(crypto_factory))


cdef shared_ptr[CKmsConnectionConfig] pyarrow_unwrap_kmsconnectionconfig(object kmsconnectionconfig) except *:
    if isinstance(kmsconnectionconfig, KmsConnectionConfig):
        return (<KmsConnectionConfig> kmsconnectionconfig).unwrap()
    raise TypeError("Expected KmsConnectionConfig, got %s" % type(kmsconnectionconfig))


cdef shared_ptr[CEncryptionConfiguration] pyarrow_unwrap_encryptionconfig(object encryptionconfig) except *:
    if isinstance(encryptionconfig, EncryptionConfiguration):
        return (<EncryptionConfiguration> encryptionconfig).unwrap()
    raise TypeError("Expected EncryptionConfiguration, got %s" % type(encryptionconfig))


cdef shared_ptr[CDecryptionConfiguration] pyarrow_unwrap_decryptionconfig(object decryptionconfig) except *:
    if isinstance(decryptionconfig, DecryptionConfiguration):
        return (<DecryptionConfiguration> decryptionconfig).unwrap()
    raise TypeError("Expected DecryptionConfiguration, got %s" % type(decryptionconfig))


def create_decryption_properties(
    footer_key,
    *,
    aad_prefix=None,
    bint check_footer_integrity=True,
    bint allow_plaintext_files=False,
):
    """
    Create FileDecryptionProperties using a direct footer key.

    This bypasses the KMS-based :class:`CryptoFactory` API and directly
    constructs decryption properties from a plaintext key. This is useful
    when the caller manages key wrapping externally (e.g. via an
    application-level envelope encryption scheme).

    For most use cases, prefer the higher-level :class:`CryptoFactory`
    with :class:`DecryptionConfiguration`, which handles envelope
    encryption and key rotation automatically.

    Parameters
    ----------
    footer_key : bytes
        The decryption key for the file footer (and all columns if
        uniform encryption was used). Must be 16, 24, or 32 bytes
        for AES-128, AES-192, or AES-256 respectively.
    aad_prefix : bytes, optional
        Additional Authenticated Data prefix. Must match the AAD prefix
        that was used during encryption. Required if the file was written
        with ``store_aad_prefix=False``.
    check_footer_integrity : bool, default True
        Whether to verify footer integrity using the signature stored
        in the file. Set to False only for debugging.
    allow_plaintext_files : bool, default False
        Whether to allow reading plaintext (unencrypted) files with
        these decryption properties without raising an error.

    Returns
    -------
    FileDecryptionProperties
        Properties that can be passed to :func:`read_table`,
        :class:`ParquetFile`, or
        :class:`~pyarrow.dataset.ParquetFragmentScanOptions`.

    Examples
    --------
    >>> import pyarrow.parquet as pq
    >>> import pyarrow.parquet.encryption as pe
    >>> props = pe.create_decryption_properties(
    ...     footer_key=b'0123456789abcdef',
    ...     aad_prefix=b'table_id'
    ... )
    >>> table = pq.read_table('encrypted.parquet', decryption_properties=props)
    """
    cdef:
        CSecureString c_footer_key
        c_string c_aad_prefix
        CFileDecryptionPropertiesBuilder* builder
        shared_ptr[CFileDecryptionProperties] props

    footer_key_bytes = tobytes(footer_key)
    if len(footer_key_bytes) not in (16, 24, 32):
        raise ValueError(
            f"footer_key must be 16, 24, or 32 bytes, got {len(footer_key_bytes)}"
        )

    c_footer_key = CSecureString(<c_string>footer_key_bytes)
    builder = new CFileDecryptionPropertiesBuilder()

    try:
        builder.footer_key(c_footer_key)

        if aad_prefix is not None:
            c_aad_prefix = tobytes(aad_prefix)
            builder.aad_prefix(c_aad_prefix)

        if not check_footer_integrity:
            builder.disable_footer_signature_verification()

        if allow_plaintext_files:
            builder.plaintext_files_allowed()

        props = builder.build()
    finally:
        del builder

    return FileDecryptionProperties.wrap(props)


def create_encryption_properties(
    footer_key,
    *,
    aad_prefix=None,
    bint store_aad_prefix=True,
    encryption_algorithm="AES_GCM_V1",
    bint plaintext_footer=False,
):
    """
    Create FileEncryptionProperties using a direct footer key.

    This bypasses the KMS-based :class:`CryptoFactory` API and directly
    constructs encryption properties from a plaintext key. This is useful
    when the caller manages key wrapping externally (e.g. via an
    application-level envelope encryption scheme).

    For most use cases, prefer the higher-level :class:`CryptoFactory`
    with :class:`EncryptionConfiguration`, which handles envelope
    encryption, key rotation, and unique-per-file data keys
    automatically.

    Parameters
    ----------
    footer_key : bytes
        The encryption key for the file footer (and all columns unless
        per-column keys are specified). Must be 16, 24, or 32 bytes
        for AES-128, AES-192, or AES-256 respectively.
    aad_prefix : bytes, optional
        Additional Authenticated Data prefix for cryptographic binding.
    store_aad_prefix : bool, default True
        Whether to store the AAD prefix in the Parquet file metadata.
        Set to False when the AAD prefix will be supplied externally
        at read time.
        Only meaningful when *aad_prefix* is provided.
    encryption_algorithm : str, default "AES_GCM_V1"
        Encryption algorithm. Either ``"AES_GCM_V1"`` or
        ``"AES_GCM_CTR_V1"``.
    plaintext_footer : bool, default False
        Whether to leave the file footer unencrypted. When True, file
        schema and column statistics are readable without a key.

    Returns
    -------
    FileEncryptionProperties
        Properties that can be passed to :func:`write_table` or
        :class:`ParquetWriter`.

    Examples
    --------
    >>> import pyarrow as pa
    >>> import pyarrow.parquet as pq
    >>> import pyarrow.parquet.encryption as pe
    >>> props = pe.create_encryption_properties(
    ...     footer_key=b'0123456789abcdef',
    ...     aad_prefix=b'table_id',
    ...     store_aad_prefix=False
    ... )
    >>> pq.write_table(table, 'encrypted.parquet', encryption_properties=props)
    """
    cdef:
        CSecureString c_footer_key
        c_string c_aad_prefix
        CFileEncryptionPropertiesBuilder* builder
        shared_ptr[CFileEncryptionProperties] props
        ParquetCipher cipher

    footer_key_bytes = tobytes(footer_key)
    if len(footer_key_bytes) not in (16, 24, 32):
        raise ValueError(
            f"footer_key must be 16, 24, or 32 bytes, got {len(footer_key_bytes)}"
        )

    cipher = cipher_from_name(encryption_algorithm)
    c_footer_key = CSecureString(<c_string>footer_key_bytes)
    builder = new CFileEncryptionPropertiesBuilder(c_footer_key)

    try:
        builder.algorithm(cipher)

        if aad_prefix is not None:
            c_aad_prefix = tobytes(aad_prefix)
            builder.aad_prefix(c_aad_prefix)
            if not store_aad_prefix:
                builder.disable_aad_prefix_storage()

        if plaintext_footer:
            builder.set_plaintext_footer()

        props = builder.build()
    finally:
        del builder

    return FileEncryptionProperties.wrap(props)