diff --git a/cuenca_validations/types/__init__.py b/cuenca_validations/types/__init__.py index b5eb1ba2..2912efc8 100644 --- a/cuenca_validations/types/__init__.py +++ b/cuenca_validations/types/__init__.py @@ -104,6 +104,7 @@ 'digits', 'get_state_name', 'uuid_field', + 'LogConfig', ] from .card import StrictPaymentCardNumber @@ -152,6 +153,7 @@ from .files import BatchFileMetadata from .general import ( JSONEncoder, + LogConfig, SantizedDict, StrictPositiveInt, digits, diff --git a/cuenca_validations/types/general.py b/cuenca_validations/types/general.py index 52f9cf09..d00132b3 100644 --- a/cuenca_validations/types/general.py +++ b/cuenca_validations/types/general.py @@ -1,4 +1,5 @@ import json +from dataclasses import dataclass from typing import Annotated, Any, Optional from pydantic import AnyUrl, Field, HttpUrl, PlainSerializer, StringConstraints @@ -82,3 +83,10 @@ def digits( def get_state_name(state: State): return names_state[state] + + +@dataclass +class LogConfig: + masked: bool = False + unmasked_chars_length: int = 0 + excluded: bool = False diff --git a/cuenca_validations/types/helpers.py b/cuenca_validations/types/helpers.py index 8e9d1f92..23a80c02 100644 --- a/cuenca_validations/types/helpers.py +++ b/cuenca_validations/types/helpers.py @@ -1,6 +1,10 @@ import uuid from base64 import urlsafe_b64encode -from typing import Callable +from typing import Callable, Optional + +from pydantic.fields import FieldInfo + +from .general import LogConfig def uuid_field(prefix: str = '') -> Callable[[], str]: @@ -8,3 +12,11 @@ def base64_uuid_func() -> str: return prefix + urlsafe_b64encode(uuid.uuid4().bytes).decode()[:-2] return base64_uuid_func + + +def get_log_config(field: FieldInfo) -> Optional[LogConfig]: + """Helper function to find LogConfig in field metadata""" + try: + return next(m for m in field.metadata if isinstance(m, LogConfig)) + except StopIteration: + return None diff --git a/cuenca_validations/types/requests.py b/cuenca_validations/types/requests.py index b0cdee98..69831387 100644 --- a/cuenca_validations/types/requests.py +++ b/cuenca_validations/types/requests.py @@ -54,7 +54,12 @@ PaymentCardNumber, StrictPaymentCardNumber, ) -from .general import SerializableAnyUrl, SerializableHttpUrl, StrictPositiveInt +from .general import ( + LogConfig, + SerializableAnyUrl, + SerializableHttpUrl, + StrictPositiveInt, +) from .identities import ( Address, Beneficiary, @@ -490,7 +495,9 @@ def beneficiary_percentage( class UserLoginRequest(BaseRequest): - password: str # Set password field to str for backward compatibility. + password: Annotated[ + str, LogConfig(masked=True) + ] # Set password field to str for backward compatibility. user_id: Optional[str] = Field(None, description='Deprecated field') model_config = ConfigDict( json_schema_extra={'example': {'password': 'supersecret'}}, diff --git a/cuenca_validations/version.py b/cuenca_validations/version.py index f593cd5b..a33997dd 100644 --- a/cuenca_validations/version.py +++ b/cuenca_validations/version.py @@ -1 +1 @@ -__version__ = '2.0.4' +__version__ = '2.1.0' diff --git a/tests/test_types.py b/tests/test_types.py index 39322e7a..305df9ec 100644 --- a/tests/test_types.py +++ b/tests/test_types.py @@ -2,10 +2,12 @@ import json from dataclasses import dataclass from enum import Enum +from typing import Annotated import pytest from freezegun import freeze_time -from pydantic import BaseModel, SecretStr, ValidationError +from pydantic import AfterValidator, BaseModel, SecretStr, ValidationError +from pydantic.fields import FieldInfo from cuenca_validations.types import ( Address, @@ -24,7 +26,9 @@ SessionType, State, ) -from cuenca_validations.types.general import StrictPositiveInt +from cuenca_validations.types.general import LogConfig, StrictPositiveInt +from cuenca_validations.types.helpers import get_log_config +from cuenca_validations.types.identities import Password from cuenca_validations.types.requests import ( ApiKeyUpdateRequest, BankAccountValidationRequest, @@ -586,3 +590,63 @@ class IntModel(BaseModel): def test_strict_positive_int_invalid(value, expected_error, expected_message): with pytest.raises(expected_error, match=expected_message): IntModel(value=value) + + +def validate_repeated_digits(password: str) -> str: + """ + Example of a custom validator + Check if the str contains repeated numbers + """ + import re + + if re.search(r'(\d).*\1', password): + raise ValueError("str cannot contain repeated digits") + return password + + +class LogConfigModel(BaseModel): + password: Annotated[Password, LogConfig(masked=True)] + validated: Annotated[ + str, AfterValidator(validate_repeated_digits), LogConfig(masked=True) + ] + secret: Annotated[str, LogConfig(masked=True)] + partial_secret: Annotated[ + str, LogConfig(masked=True, unmasked_chars_length=4) + ] + unmasked: Annotated[str, LogConfig(masked=False)] + excluded: Annotated[str, LogConfig(excluded=True)] + + +@pytest.mark.parametrize( + "field_name,expected_masked,expected_unmasked_length,expected_excluded", + [ + ("password", True, 0, False), + ("validated", True, 0, False), + ("secret", True, 0, False), + ("partial_secret", True, 4, False), + ("unmasked", False, 0, False), + ("excluded", False, 0, True), + ], +) +def test_log_config( + field_name, expected_masked, expected_unmasked_length, expected_excluded +): + model = LogConfigModel( + password="Mypass123.", + validated="str123", + secret="super-secret", + partial_secret="1234567890", + unmasked="unmasked", + excluded="excluded", + ) + + field = model.model_fields[field_name] + config = get_log_config(field) + assert config.masked is expected_masked + assert config.unmasked_chars_length == expected_unmasked_length + assert config.excluded is expected_excluded + + +def test_get_log_config_no_log_config(): + field = FieldInfo(default=None) + assert get_log_config(field) is None