Skip to content

Provider

MODULE DESCRIPTION
base
kms
CLASS DESCRIPTION
BaseProvider

The BaseProvider class serves as an abstract base class for data masking providers.

BaseProvider 

BaseProvider(
    json_serializer: Callable[..., str] = functools.partial(
        json.dumps, ensure_ascii=False
    ),
    json_deserializer: Callable[[str], Any] = json.loads,
)

The BaseProvider class serves as an abstract base class for data masking providers.

Example 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from aws_lambda_powertools.utilities._data_masking.provider import BaseProvider
from aws_lambda_powertools.utilities.data_masking import DataMasking

class MyCustomProvider(BaseProvider):
    def encrypt(self, data) -> str:
        # Implementation logic for data encryption

    def decrypt(self, data) -> Any:
        # Implementation logic for data decryption

    def erase(self, data) -> str | Iterable:
        # Implementation logic for data masking
        pass

def lambda_handler(event, context):
    provider = MyCustomProvider(["secret-key"])
    data_masker = DataMasking(provider=provider)

    data = {
        "project": "powertools",
        "sensitive": "password"
    }

    encrypted = data_masker.encrypt(data)

    return encrypted
METHOD DESCRIPTION
decrypt

Abstract method for decrypting data. Subclasses must implement this method.
encrypt

Abstract method for encrypting data. Subclasses must implement this method.
erase

This method irreversibly erases data.
Source code in aws_lambda_powertools/utilities/data_masking/provider/base.py 
46
47
48
49
50
51
52
def __init__(
    self,
    json_serializer: Callable[..., str] = functools.partial(json.dumps, ensure_ascii=False),
    json_deserializer: Callable[[str], Any] = json.loads,
) -> None:
    self.json_serializer = json_serializer
    self.json_deserializer = json_deserializer

decrypt 

decrypt(
    data,
    provider_options: dict | None = None,
    **encryption_context: str
) -> Any

Abstract method for decrypting data. Subclasses must implement this method.

Source code in aws_lambda_powertools/utilities/data_masking/provider/base.py 
60
61
62
63
64
def decrypt(self, data, provider_options: dict | None = None, **encryption_context: str) -> Any:
    """
    Abstract method for decrypting data. Subclasses must implement this method.
    """
    raise NotImplementedError("Subclasses must implement decrypt()")

encrypt 

encrypt(
    data,
    provider_options: dict | None = None,
    **encryption_context: str
) -> str

Abstract method for encrypting data. Subclasses must implement this method.

Source code in aws_lambda_powertools/utilities/data_masking/provider/base.py 
54
55
56
57
58
def encrypt(self, data, provider_options: dict | None = None, **encryption_context: str) -> str:
    """
    Abstract method for encrypting data. Subclasses must implement this method.
    """
    raise NotImplementedError("Subclasses must implement encrypt()")

erase 

erase(data, **kwargs) -> Iterable[str]

This method irreversibly erases data.

If the data to be erased is of type str, dict, or bytes, this method will return an erased string, i.e. "*".

If the data to be erased is of an iterable type like list, tuple, or set, this method will return a new object of the same type as the input data but with each element replaced by the string "*".

Source code in aws_lambda_powertools/utilities/data_masking/provider/base.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def erase(self, data, **kwargs) -> Iterable[str]:
    """
    This method irreversibly erases data.

    If the data to be erased is of type `str`, `dict`, or `bytes`,
    this method will return an erased string, i.e. "*****".

    If the data to be erased is of an iterable type like `list`, `tuple`,
    or `set`, this method will return a new object of the same type as the
    input data but with each element replaced by the string "*****".
    """
    if isinstance(data, (str, dict, bytes)):
        return DATA_MASKING_STRING
    elif isinstance(data, (list, tuple, set)):
        return type(data)([DATA_MASKING_STRING] * len(data))
    return DATA_MASKING_STRING

base

CLASS DESCRIPTION
BaseProvider

The BaseProvider class serves as an abstract base class for data masking providers.

BaseProvider 

BaseProvider(
    json_serializer: Callable[..., str] = functools.partial(
        json.dumps, ensure_ascii=False
    ),
    json_deserializer: Callable[[str], Any] = json.loads,
)

The BaseProvider class serves as an abstract base class for data masking providers.

Example 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from aws_lambda_powertools.utilities._data_masking.provider import BaseProvider
from aws_lambda_powertools.utilities.data_masking import DataMasking

class MyCustomProvider(BaseProvider):
    def encrypt(self, data) -> str:
        # Implementation logic for data encryption

    def decrypt(self, data) -> Any:
        # Implementation logic for data decryption

    def erase(self, data) -> str | Iterable:
        # Implementation logic for data masking
        pass

def lambda_handler(event, context):
    provider = MyCustomProvider(["secret-key"])
    data_masker = DataMasking(provider=provider)

    data = {
        "project": "powertools",
        "sensitive": "password"
    }

    encrypted = data_masker.encrypt(data)

    return encrypted
METHOD DESCRIPTION
decrypt

Abstract method for decrypting data. Subclasses must implement this method.
encrypt

Abstract method for encrypting data. Subclasses must implement this method.
erase

This method irreversibly erases data.
Source code in aws_lambda_powertools/utilities/data_masking/provider/base.py 
46
47
48
49
50
51
52
def __init__(
    self,
    json_serializer: Callable[..., str] = functools.partial(json.dumps, ensure_ascii=False),
    json_deserializer: Callable[[str], Any] = json.loads,
) -> None:
    self.json_serializer = json_serializer
    self.json_deserializer = json_deserializer

decrypt 

decrypt(
    data,
    provider_options: dict | None = None,
    **encryption_context: str
) -> Any

Abstract method for decrypting data. Subclasses must implement this method.

Source code in aws_lambda_powertools/utilities/data_masking/provider/base.py 
60
61
62
63
64
def decrypt(self, data, provider_options: dict | None = None, **encryption_context: str) -> Any:
    """
    Abstract method for decrypting data. Subclasses must implement this method.
    """
    raise NotImplementedError("Subclasses must implement decrypt()")

encrypt 

encrypt(
    data,
    provider_options: dict | None = None,
    **encryption_context: str
) -> str

Abstract method for encrypting data. Subclasses must implement this method.

Source code in aws_lambda_powertools/utilities/data_masking/provider/base.py 
54
55
56
57
58
def encrypt(self, data, provider_options: dict | None = None, **encryption_context: str) -> str:
    """
    Abstract method for encrypting data. Subclasses must implement this method.
    """
    raise NotImplementedError("Subclasses must implement encrypt()")

erase 

erase(data, **kwargs) -> Iterable[str]

This method irreversibly erases data.

If the data to be erased is of type str, dict, or bytes, this method will return an erased string, i.e. "*".

If the data to be erased is of an iterable type like list, tuple, or set, this method will return a new object of the same type as the input data but with each element replaced by the string "*".

Source code in aws_lambda_powertools/utilities/data_masking/provider/base.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def erase(self, data, **kwargs) -> Iterable[str]:
    """
    This method irreversibly erases data.

    If the data to be erased is of type `str`, `dict`, or `bytes`,
    this method will return an erased string, i.e. "*****".

    If the data to be erased is of an iterable type like `list`, `tuple`,
    or `set`, this method will return a new object of the same type as the
    input data but with each element replaced by the string "*****".
    """
    if isinstance(data, (str, dict, bytes)):
        return DATA_MASKING_STRING
    elif isinstance(data, (list, tuple, set)):
        return type(data)([DATA_MASKING_STRING] * len(data))
    return DATA_MASKING_STRING

kms

MODULE DESCRIPTION
aws_encryption_sdk
CLASS DESCRIPTION
AWSEncryptionSDKProvider

The AWSEncryptionSDKProvider is used as a provider for the DataMasking class.

AWSEncryptionSDKProvider 

AWSEncryptionSDKProvider(
    keys: list[str],
    key_provider=None,
    local_cache_capacity: int = CACHE_CAPACITY,
    max_cache_age_seconds: float = MAX_CACHE_AGE_SECONDS,
    max_messages_encrypted: int = MAX_MESSAGES_ENCRYPTED,
    max_bytes_encrypted: int = MAX_BYTES_ENCRYPTED,
    json_serializer: Callable[..., str] = functools.partial(
        json.dumps, ensure_ascii=False
    ),
    json_deserializer: Callable[[str], Any] = json.loads,
)

Bases: BaseProvider

The AWSEncryptionSDKProvider is used as a provider for the DataMasking class.

Example 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
from aws_lambda_powertools.utilities.data_masking import DataMasking
from aws_lambda_powertools.utilities.data_masking.providers.kms.aws_encryption_sdk import (
    AWSEncryptionSDKProvider,
)


def lambda_handler(event, context):
    provider = AWSEncryptionSDKProvider(["arn:aws:kms:us-east-1:0123456789012:key/key-id"])
    data_masker = DataMasking(provider=provider)

    data = {
        "project": "powertools",
        "sensitive": "password"
    }

    encrypted = data_masker.encrypt(data)

    return encrypted
Source code in aws_lambda_powertools/utilities/data_masking/provider/kms/aws_encryption_sdk.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def __init__(
    self,
    keys: list[str],
    key_provider=None,
    local_cache_capacity: int = CACHE_CAPACITY,
    max_cache_age_seconds: float = MAX_CACHE_AGE_SECONDS,
    max_messages_encrypted: int = MAX_MESSAGES_ENCRYPTED,
    max_bytes_encrypted: int = MAX_BYTES_ENCRYPTED,
    json_serializer: Callable[..., str] = functools.partial(json.dumps, ensure_ascii=False),
    json_deserializer: Callable[[str], Any] = json.loads,
):
    super().__init__(json_serializer=json_serializer, json_deserializer=json_deserializer)

    self._key_provider = key_provider or KMSKeyProvider(
        keys=keys,
        local_cache_capacity=local_cache_capacity,
        max_cache_age_seconds=max_cache_age_seconds,
        max_messages_encrypted=max_messages_encrypted,
        max_bytes_encrypted=max_bytes_encrypted,
        json_serializer=json_serializer,
        json_deserializer=json_deserializer,
    )

aws_encryption_sdk

CLASS DESCRIPTION
AWSEncryptionSDKProvider

The AWSEncryptionSDKProvider is used as a provider for the DataMasking class.
KMSKeyProvider

The KMSKeyProvider is responsible for assembling an AWS Key Management Service (KMS)

AWSEncryptionSDKProvider 

AWSEncryptionSDKProvider(
    keys: list[str],
    key_provider=None,
    local_cache_capacity: int = CACHE_CAPACITY,
    max_cache_age_seconds: float = MAX_CACHE_AGE_SECONDS,
    max_messages_encrypted: int = MAX_MESSAGES_ENCRYPTED,
    max_bytes_encrypted: int = MAX_BYTES_ENCRYPTED,
    json_serializer: Callable[..., str] = functools.partial(
        json.dumps, ensure_ascii=False
    ),
    json_deserializer: Callable[[str], Any] = json.loads,
)

Bases: BaseProvider

The AWSEncryptionSDKProvider is used as a provider for the DataMasking class.

Example 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
from aws_lambda_powertools.utilities.data_masking import DataMasking
from aws_lambda_powertools.utilities.data_masking.providers.kms.aws_encryption_sdk import (
    AWSEncryptionSDKProvider,
)


def lambda_handler(event, context):
    provider = AWSEncryptionSDKProvider(["arn:aws:kms:us-east-1:0123456789012:key/key-id"])
    data_masker = DataMasking(provider=provider)

    data = {
        "project": "powertools",
        "sensitive": "password"
    }

    encrypted = data_masker.encrypt(data)

    return encrypted
Source code in aws_lambda_powertools/utilities/data_masking/provider/kms/aws_encryption_sdk.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def __init__(
    self,
    keys: list[str],
    key_provider=None,
    local_cache_capacity: int = CACHE_CAPACITY,
    max_cache_age_seconds: float = MAX_CACHE_AGE_SECONDS,
    max_messages_encrypted: int = MAX_MESSAGES_ENCRYPTED,
    max_bytes_encrypted: int = MAX_BYTES_ENCRYPTED,
    json_serializer: Callable[..., str] = functools.partial(json.dumps, ensure_ascii=False),
    json_deserializer: Callable[[str], Any] = json.loads,
):
    super().__init__(json_serializer=json_serializer, json_deserializer=json_deserializer)

    self._key_provider = key_provider or KMSKeyProvider(
        keys=keys,
        local_cache_capacity=local_cache_capacity,
        max_cache_age_seconds=max_cache_age_seconds,
        max_messages_encrypted=max_messages_encrypted,
        max_bytes_encrypted=max_bytes_encrypted,
        json_serializer=json_serializer,
        json_deserializer=json_deserializer,
    )

KMSKeyProvider 

KMSKeyProvider(
    keys: list[str],
    json_serializer: Callable[..., str],
    json_deserializer: Callable[[str], Any],
    local_cache_capacity: int = CACHE_CAPACITY,
    max_cache_age_seconds: float = MAX_CACHE_AGE_SECONDS,
    max_messages_encrypted: int = MAX_MESSAGES_ENCRYPTED,
    max_bytes_encrypted: int = MAX_BYTES_ENCRYPTED,
)

The KMSKeyProvider is responsible for assembling an AWS Key Management Service (KMS) client, a caching mechanism, and a keyring for secure key management and data encryption.

METHOD DESCRIPTION
decrypt

Decrypt data using AWSEncryptionSDKProvider.
encrypt

Encrypt data using the AWSEncryptionSDKProvider.
Source code in aws_lambda_powertools/utilities/data_masking/provider/kms/aws_encryption_sdk.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def __init__(
    self,
    keys: list[str],
    json_serializer: Callable[..., str],
    json_deserializer: Callable[[str], Any],
    local_cache_capacity: int = CACHE_CAPACITY,
    max_cache_age_seconds: float = MAX_CACHE_AGE_SECONDS,
    max_messages_encrypted: int = MAX_MESSAGES_ENCRYPTED,
    max_bytes_encrypted: int = MAX_BYTES_ENCRYPTED,
):
    session = botocore.session.Session()
    register_feature_to_botocore_session(session, "data-masking")

    self.json_serializer = json_serializer
    self.json_deserializer = json_deserializer
    self.client = EncryptionSDKClient()
    self.keys = keys
    self.cache = LocalCryptoMaterialsCache(local_cache_capacity)
    self.key_provider = StrictAwsKmsMasterKeyProvider(key_ids=self.keys, botocore_session=session)
    self.cache_cmm = CachingCryptoMaterialsManager(
        master_key_provider=self.key_provider,
        cache=self.cache,
        max_age=max_cache_age_seconds,
        max_messages_encrypted=max_messages_encrypted,
        max_bytes_encrypted=max_bytes_encrypted,
    )
decrypt 
decrypt(
    data: str,
    provider_options: dict | None = None,
    **encryption_context: str
) -> Any

Decrypt data using AWSEncryptionSDKProvider.

PARAMETER DESCRIPTION
data

The encrypted data, as a base64-encoded string

TYPE: str

provider_options

Additional options for the aws_encryption_sdk.EncryptionSDKClient

TYPE: dict | None DEFAULT: None

RETURNS DESCRIPTION
ciphertext

The decrypted data in bytes

TYPE: bytes

Source code in aws_lambda_powertools/utilities/data_masking/provider/kms/aws_encryption_sdk.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
def decrypt(self, data: str, provider_options: dict | None = None, **encryption_context: str) -> Any:
    """
    Decrypt data using AWSEncryptionSDKProvider.

    Parameters
    -------
    data: str
        The encrypted data, as a base64-encoded string
    provider_options
        Additional options for the aws_encryption_sdk.EncryptionSDKClient

    Returns
    -------
    ciphertext: bytes
        The decrypted data in bytes
    """
    provider_options = provider_options or {}
    self._validate_encryption_context(encryption_context)

    try:
        ciphertext_decoded = base64_decode(data)
    except Error:
        raise DataMaskingDecryptValueError(
            "Data decryption failed. Please ensure that you are attempting to decrypt data that was previously encrypted.",  # noqa E501
        )

    try:
        ciphertext, decryptor_header = self.client.decrypt(
            source=ciphertext_decoded,
            key_provider=self.key_provider,
            **provider_options,
        )
    except DecryptKeyError:
        raise DataMaskingDecryptKeyError(
            "Failed to decrypt data - Please ensure you are using a valid Symmetric AWS KMS Key ARN, not KMS Key ID or alias.",  # noqa E501
        )
    except (TypeError, NotSupportedError):
        raise DataMaskingDecryptValueError(
            "Data decryption failed. Please ensure that you are attempting to decrypt data that was previously encrypted.",  # noqa E501
        )

    self._compare_encryption_context(decryptor_header.encryption_context, encryption_context)

    decoded_ciphertext = bytes_to_string(ciphertext)

    return self.json_deserializer(decoded_ciphertext)
encrypt 
encrypt(
    data: Any,
    provider_options: dict | None = None,
    **encryption_context: str
) -> str

Encrypt data using the AWSEncryptionSDKProvider.

PARAMETER DESCRIPTION
data

The data to be encrypted.

TYPE: Any

provider_options

Additional options for the aws_encryption_sdk.EncryptionSDKClient

TYPE: dict | None DEFAULT: None

**encryption_context

Additional keyword arguments collected into a dictionary.

TYPE: str DEFAULT: {}

RETURNS DESCRIPTION
ciphertext

The encrypted data, as a base64-encoded string.

TYPE: str

Source code in aws_lambda_powertools/utilities/data_masking/provider/kms/aws_encryption_sdk.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def encrypt(self, data: Any, provider_options: dict | None = None, **encryption_context: str) -> str:
    """
    Encrypt data using the AWSEncryptionSDKProvider.

    Parameters
    -------
    data: Any
        The data to be encrypted.
    provider_options: dict
        Additional options for the aws_encryption_sdk.EncryptionSDKClient
    **encryption_context: str
        Additional keyword arguments collected into a dictionary.

    Returns
    -------
    ciphertext: str
        The encrypted data, as a base64-encoded string.
    """
    provider_options = provider_options or {}
    self._validate_encryption_context(encryption_context)

    data_encoded = self.json_serializer(data).encode("utf-8")

    try:
        ciphertext, _ = self.client.encrypt(
            source=data_encoded,
            materials_manager=self.cache_cmm,
            encryption_context=encryption_context,
            **provider_options,
        )
    except GenerateKeyError:
        raise DataMaskingEncryptKeyError(
            "Failed to encrypt data. Please ensure you are using a valid Symmetric AWS KMS Key ARN, not KMS Key ID or alias.",  # noqa E501
        )

    return bytes_to_base64_string(ciphertext)