| # 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 is a low-level API that constructs decryption properties directly |
| from a plaintext key, bypassing the KMS-based :class:`CryptoFactory`. |
| It is intended for callers that manage key wrapping and storage |
| themselves (e.g. an application-level scheme). |
| |
| For most use cases, prefer the higher-level :class:`CryptoFactory` |
| with :class:`DecryptionConfiguration`, which implements the full |
| Parquet key management specification and is interoperable with |
| other tools and frameworks. |
| |
| .. note:: |
| Currently only uniform encryption (single key for footer and all |
| columns) is supported with this method. Per-column keys are not |
| yet available; files encrypted with per-column keys cannot be |
| decrypted using this function. |
| |
| Parameters |
| ---------- |
| footer_key : bytes |
| The decryption key for the file footer and all columns (uniform |
| encryption). 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 AAD prefix was |
| not stored in the file metadata during encryption. |
| 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:`~pyarrow.parquet.read_table`, |
| :class:`~pyarrow.parquet.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: |
| c_string c_footer_key_str |
| CSecureString c_footer_key |
| CFileDecryptionPropertiesBuilder builder |
| shared_ptr[CFileDecryptionProperties] props |
| |
| if not isinstance(footer_key, bytes): |
| raise TypeError( |
| f"footer_key must be bytes, not {type(footer_key).__name__}" |
| ) |
| if len(footer_key) not in (16, 24, 32): |
| raise ValueError( |
| f"footer_key must be 16, 24, or 32 bytes, got {len(footer_key)}" |
| ) |
| |
| c_footer_key_str = <c_string>footer_key |
| c_footer_key = CSecureString(move(c_footer_key_str)) |
| builder.footer_key(c_footer_key) |
| |
| if aad_prefix is not None: |
| if not isinstance(aad_prefix, bytes): |
| raise TypeError( |
| f"aad_prefix must be bytes, not {type(aad_prefix).__name__}" |
| ) |
| builder.aad_prefix(<c_string>aad_prefix) |
| |
| if not check_footer_integrity: |
| builder.disable_footer_signature_verification() |
| |
| if allow_plaintext_files: |
| builder.plaintext_files_allowed() |
| |
| props = builder.build() |
| |
| 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 is a low-level API that constructs encryption properties directly |
| from a plaintext key, bypassing the KMS-based :class:`CryptoFactory`. |
| It is intended for callers that manage key wrapping and storage |
| themselves (e.g. an application-level scheme). |
| |
| .. warning:: |
| The caller is responsible for key management best practices. |
| Reusing the same key for multiple files without unique data keys |
| weakens AES-GCM security. The higher-level :class:`CryptoFactory` |
| with :class:`EncryptionConfiguration` handles this automatically |
| and is interoperable with other tools and frameworks -- |
| prefer it unless you have a specific reason to manage |
| keys yourself. |
| |
| .. note:: |
| Currently only uniform encryption (single key for footer and all |
| columns) is supported with this method. Per-column keys are not |
| yet available; the provided key encrypts both the footer and |
| every column. |
| |
| Parameters |
| ---------- |
| footer_key : bytes |
| The encryption key for the file footer and all columns (uniform |
| encryption). 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 metadata are readable without a key. |
| |
| Returns |
| ------- |
| FileEncryptionProperties |
| Properties that can be passed to :func:`~pyarrow.parquet.write_table` or |
| :class:`~pyarrow.parquet.ParquetWriter`. |
| |
| Examples |
| -------- |
| >>> import pyarrow as pa |
| >>> import pyarrow.parquet as pq |
| >>> import pyarrow.parquet.encryption as pe |
| >>> table = pa.table({'col': [1, 2, 3]}) |
| >>> 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: |
| c_string c_footer_key_str |
| CSecureString c_footer_key |
| CFileEncryptionPropertiesBuilder* builder |
| shared_ptr[CFileEncryptionProperties] props |
| ParquetCipher cipher |
| |
| if not isinstance(footer_key, bytes): |
| raise TypeError( |
| f"footer_key must be bytes, not {type(footer_key).__name__}" |
| ) |
| if len(footer_key) not in (16, 24, 32): |
| raise ValueError( |
| f"footer_key must be 16, 24, or 32 bytes, got {len(footer_key)}" |
| ) |
| |
| cipher = cipher_from_name(encryption_algorithm) |
| c_footer_key_str = <c_string>footer_key |
| c_footer_key = CSecureString(move(c_footer_key_str)) |
| builder = new CFileEncryptionPropertiesBuilder(c_footer_key) |
| try: |
| builder.algorithm(cipher) |
| |
| if aad_prefix is not None: |
| if not isinstance(aad_prefix, bytes): |
| raise TypeError( |
| f"aad_prefix must be bytes, not {type(aad_prefix).__name__}" |
| ) |
| builder.aad_prefix(<c_string>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) |
