from dataclasses import dataclass import itsdangerous from fansifter_common.utils.functional import lazy_proxy from email_campaigns.config import settings from email_campaigns.emails.enums import TokenDecryptionStatus _serializer = lazy_proxy( lambda: itsdangerous.URLSafeTimedSerializer( settings.email_public_preview_secret_key ) ) @dataclass class TokenDecryptionResult: """Result of token decryption.""" status: TokenDecryptionStatus email_id: str | None def encrypt_email_id(email_id: str) -> str: """Generate URL-safe signed token with timestamp.""" return _serializer.dumps(email_id) def decrypt_token(token: str) -> TokenDecryptionResult: """Verify and decode public preview token to get email ID. Returns TokenDecryptionResult with status: - VALID: Token is valid and not expired - EXPIRED: Token signature is valid but TTL has expired (legitimate old link) - INVALID: Token signature is invalid or malformed (possible attack) """ try: email_id = _serializer.loads( token, max_age=settings.email_public_preview_ttl_seconds ) return TokenDecryptionResult( status=TokenDecryptionStatus.VALID, email_id=email_id, ) except itsdangerous.SignatureExpired as e: try: if e.payload is None: raise ValueError("Payload is None") email_id = _serializer.load_payload(e.payload) return TokenDecryptionResult( status=TokenDecryptionStatus.EXPIRED, email_id=email_id, ) except Exception: return TokenDecryptionResult( status=TokenDecryptionStatus.INVALID, email_id=None, ) except itsdangerous.BadSignature, itsdangerous.BadData: return TokenDecryptionResult( status=TokenDecryptionStatus.INVALID, email_id=None, ) except Exception: return TokenDecryptionResult( status=TokenDecryptionStatus.INVALID, email_id=None, )