92 lines
3.0 KiB
Python
92 lines
3.0 KiB
Python
import re
|
|
import dns.resolver
|
|
from typing import Dict, List
|
|
from common.log.module_logger import ModuleLogger
|
|
from common.exception.exceptions import InvalidDataError
|
|
|
|
|
|
class EmailValidationHandler:
|
|
def __init__(self):
|
|
self.module_logger = ModuleLogger(sender_id="EmailValidationHandler")
|
|
|
|
self.email_pattern = re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')
|
|
# common spam keywords
|
|
self.spam_keywords = [
|
|
'free', 'money', 'cash', 'winner', 'lottery', 'prize', 'urgent',
|
|
'limited time', 'act now', 'click here', 'buy now', 'discount'
|
|
]
|
|
|
|
async def is_valid_email(self, email: str) -> bool:
|
|
"""validate email format"""
|
|
try:
|
|
if not email or not isinstance(email, str):
|
|
return False
|
|
|
|
if not self.email_pattern.match(email):
|
|
return False
|
|
|
|
# length check: RFC 5321 regulate the max length of email-254
|
|
if len(email) > 254:
|
|
return False
|
|
|
|
# local part and domain part check
|
|
local_part, domain_part = email.split('@', 1)
|
|
if len(local_part) > 64 or len(local_part) == 0:
|
|
return False
|
|
if len(domain_part) > 253 or len(domain_part) == 0:
|
|
return False
|
|
|
|
# check if domain part contains valid characters
|
|
if not re.match(r'^[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', domain_part):
|
|
return False
|
|
|
|
await self.module_logger.log_info(
|
|
f"Email format validation passed: {email}",
|
|
properties={"email": email}
|
|
)
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
await self.module_logger.log_error(
|
|
f"Email validation failed: {email}",
|
|
properties={"email": email, "error": str(e)}
|
|
)
|
|
return False
|
|
|
|
async def is_valid_domain(self, email: str) -> bool:
|
|
"""check if domain part is valid"""
|
|
try:
|
|
domain = email.split('@')[1]
|
|
|
|
# check MX record
|
|
try:
|
|
mx_records = dns.resolver.resolve(domain, 'MX')
|
|
if not mx_records:
|
|
return False
|
|
except Exception:
|
|
return False
|
|
|
|
# check A record (backup)
|
|
try:
|
|
a_records = dns.resolver.resolve(domain, 'A')
|
|
if not a_records:
|
|
return False
|
|
except Exception:
|
|
pass
|
|
|
|
await self.module_logger.log_info(
|
|
f"Domain validation passed: {domain}",
|
|
properties={"domain": domain}
|
|
)
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
await self.module_logger.log_error(
|
|
f"Domain validation failed: {email}",
|
|
properties={"email": email, "error": str(e)}
|
|
)
|
|
return False
|
|
|