Skip to content

Base

Base class for Data Masking

Usage Documentation

Data masking

CLASS DESCRIPTION
DataMasking

The DataMasking class orchestrates erasing, encrypting, and decrypting

FUNCTION DESCRIPTION
prepare_data

Recursively convert complex objects into dictionaries or simple types.

DataMasking

DataMasking(
    provider: BaseProvider | None = None,
    raise_on_missing_field: bool = True,
)

The DataMasking class orchestrates erasing, encrypting, and decrypting for the base provider.

Example
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
from aws_lambda_powertools.utilities.data_masking.base import DataMasking

def lambda_handler(event, context):
    masker = DataMasking()

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

    erased = masker.erase(data,fields=["sensitive"])

    return erased
METHOD DESCRIPTION
decrypt

Decrypt data using the configured encryption provider.

encrypt

Encrypt data using the configured encryption provider.

erase

Erase or mask sensitive data in the input.

Source code in aws_lambda_powertools/utilities/data_masking/base.py
101
102
103
104
105
106
107
108
109
110
def __init__(
    self,
    provider: BaseProvider | None = None,
    raise_on_missing_field: bool = True,
):
    self.provider = provider or BaseProvider()
    # NOTE: we depend on Provider to not confuse customers in passing the same 2 serializers in 2 places
    self.json_serializer = self.provider.json_serializer
    self.json_deserializer = self.provider.json_deserializer
    self.raise_on_missing_field = raise_on_missing_field

decrypt

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

Decrypt data using the configured encryption provider.

PARAMETER DESCRIPTION
data

The data to encrypt.

TYPE: dict, Mapping, Sequence, or Number

provider_options

Provider-specific options for encryption.

TYPE: dict DEFAULT: None

**encryption_context

Additional key-value pairs for encryption context.

TYPE: str DEFAULT: {}

RETURNS DESCRIPTION
str

The encrypted data as a base64-encoded string.

Example
1
2
3
encryption_provider = AWSEncryptionSDKProvider(keys=[KMS_KEY_ARN])
data_masker = DataMasking(provider=encryption_provider)
encrypted = data_masker.decrypt(encrypted_data)
Source code in aws_lambda_powertools/utilities/data_masking/base.py
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def decrypt(
    self,
    data,
    provider_options: dict | None = None,
    **encryption_context: str,
) -> Any:
    """
    Decrypt data using the configured encryption provider.

    Parameters
    ----------
    data : dict, Mapping, Sequence, or Number
        The data to encrypt.
    provider_options : dict, optional
        Provider-specific options for encryption.
    **encryption_context : str
        Additional key-value pairs for encryption context.

    Returns
    -------
    str
        The encrypted data as a base64-encoded string.

    Example
    --------

        encryption_provider = AWSEncryptionSDKProvider(keys=[KMS_KEY_ARN])
        data_masker = DataMasking(provider=encryption_provider)
        encrypted = data_masker.decrypt(encrypted_data)
    """
    data = prepare_data(data)
    return self._apply_action(
        data=data,
        fields=None,
        action=self.provider.decrypt,
        provider_options=provider_options or {},
        dynamic_mask=None,
        custom_mask=None,
        regex_pattern=None,
        mask_format=None,
        **encryption_context,
    )

encrypt

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

Encrypt data using the configured encryption provider.

PARAMETER DESCRIPTION
data

The data to encrypt.

TYPE: dict, Mapping, Sequence, or Number

provider_options

Provider-specific options for encryption.

TYPE: dict DEFAULT: None

**encryption_context

Additional key-value pairs for encryption context.

TYPE: str DEFAULT: {}

RETURNS DESCRIPTION
str

The encrypted data as a base64-encoded string.

Example
1
2
3
encryption_provider = AWSEncryptionSDKProvider(keys=[KMS_KEY_ARN])
data_masker = DataMasking(provider=encryption_provider)
encrypted = data_masker.encrypt({"secret": "value"})
Source code in aws_lambda_powertools/utilities/data_masking/base.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
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def encrypt(
    self,
    data: dict | Mapping | Sequence | Number,
    provider_options: dict | None = None,
    **encryption_context: str,
) -> str:
    """
    Encrypt data using the configured encryption provider.

    Parameters
    ----------
    data : dict, Mapping, Sequence, or Number
        The data to encrypt.
    provider_options : dict, optional
        Provider-specific options for encryption.
    **encryption_context : str
        Additional key-value pairs for encryption context.

    Returns
    -------
    str
        The encrypted data as a base64-encoded string.

    Example
    --------

        encryption_provider = AWSEncryptionSDKProvider(keys=[KMS_KEY_ARN])
        data_masker = DataMasking(provider=encryption_provider)
        encrypted = data_masker.encrypt({"secret": "value"})
    """
    data = prepare_data(data)
    return self._apply_action(
        data=data,
        fields=None,
        action=self.provider.encrypt,
        provider_options=provider_options or {},
        dynamic_mask=None,
        custom_mask=None,
        regex_pattern=None,
        mask_format=None,
        **encryption_context,
    )

erase

erase(
    data: Any,
    fields: list[str] | None = None,
    *,
    dynamic_mask: bool | None = None,
    custom_mask: str | None = None,
    regex_pattern: str | None = None,
    mask_format: str | None = None,
    masking_rules: dict | None = None
) -> Any

Erase or mask sensitive data in the input.

PARAMETER DESCRIPTION
data

The data to be erased or masked.

TYPE: Any

fields

List of field names to be erased or masked.

TYPE: list of str DEFAULT: None

dynamic_mask

Whether to use dynamic masking.

TYPE: bool DEFAULT: None

custom_mask

Custom mask to apply instead of the default.

TYPE: str DEFAULT: None

regex_pattern

Regular expression pattern for identifying data to mask.

TYPE: str DEFAULT: None

mask_format

Format string for the mask.

TYPE: str DEFAULT: None

masking_rules

Dictionary of custom masking rules.

TYPE: dict DEFAULT: None

RETURNS DESCRIPTION
Any

The data with sensitive information erased or masked.

Source code in aws_lambda_powertools/utilities/data_masking/base.py
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
def erase(
    self,
    data: Any,
    fields: list[str] | None = None,
    *,
    dynamic_mask: bool | None = None,
    custom_mask: str | None = None,
    regex_pattern: str | None = None,
    mask_format: str | None = None,
    masking_rules: dict | None = None,
) -> Any:
    """
    Erase or mask sensitive data in the input.

    Parameters
    ----------
    data : Any
        The data to be erased or masked.
    fields : list of str, optional
        List of field names to be erased or masked.
    dynamic_mask : bool, optional
        Whether to use dynamic masking.
    custom_mask : str, optional
        Custom mask to apply instead of the default.
    regex_pattern : str, optional
        Regular expression pattern for identifying data to mask.
    mask_format : str, optional
        Format string for the mask.
    masking_rules : dict, optional
        Dictionary of custom masking rules.

    Returns
    -------
    Any
        The data with sensitive information erased or masked.
    """
    data = prepare_data(data)
    if masking_rules:
        return self._apply_masking_rules(data=data, masking_rules=masking_rules)
    else:
        return self._apply_action(
            data=data,
            fields=fields,
            action=self.provider.erase,
            dynamic_mask=dynamic_mask,
            custom_mask=custom_mask,
            regex_pattern=regex_pattern,
            mask_format=mask_format,
        )

prepare_data

prepare_data(
    data: Any, _visited: set[int] | None = None
) -> Any

Recursively convert complex objects into dictionaries or simple types. Handles dataclasses, Pydantic models, and prevents circular references.

Source code in aws_lambda_powertools/utilities/data_masking/base.py
31
32
33
34
35
36
37
38
39
40
41
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
def prepare_data(data: Any, _visited: set[int] | None = None) -> Any:
    """
    Recursively convert complex objects into dictionaries or simple types.
    Handles dataclasses, Pydantic models, and prevents circular references.
    """
    _visited = _visited or set()

    # Handle circular references and primitive types
    data_id = id(data)
    if data_id in _visited or isinstance(data, (str, int, float, bool, type(None))):
        return data

    _visited.add(data_id)

    # Define handlers as (condition, transformer) pairs
    handlers: list[tuple[Callable[[Any], bool], Callable[[Any], Any]]] = [
        # Dataclasses
        (lambda x: hasattr(x, "__dataclass_fields__"), lambda x: prepare_data(dataclasses.asdict(x), _visited)),
        # Pydantic models
        (lambda x: callable(getattr(x, "model_dump", None)), lambda x: prepare_data(x.model_dump(), _visited)),
        # Objects with dict() method
        (
            lambda x: callable(getattr(x, "dict", None)) and not isinstance(x, dict),
            lambda x: prepare_data(x.dict(), _visited),
        ),
        # Dictionaries
        (
            lambda x: isinstance(x, dict),
            lambda x: {prepare_data(k, _visited): prepare_data(v, _visited) for k, v in x.items()},
        ),
        # Lists, tuples, sets
        (lambda x: isinstance(x, (list, tuple, set)), lambda x: type(x)(prepare_data(item, _visited) for item in x)),
        # Objects with __dict__
        (lambda x: hasattr(x, "__dict__"), lambda x: prepare_data(vars(x), _visited)),
    ]

    # Find and apply the first matching handler
    for condition, transformer in handlers:
        if condition(data):
            return transformer(data)

    # Default fallback
    return data