#!/usr/bin/env python3 """ Sender blocklist checking with wildcard support """ import fnmatch from typing import List, Dict from email.utils import parseaddr from logger import log from aws.dynamodb_handler import DynamoDBHandler class BlocklistChecker: """Checks if senders are blocked""" def __init__(self, dynamodb: DynamoDBHandler): self.dynamodb = dynamodb def is_sender_blocked( self, recipient: str, sender: str, worker_name: str ) -> bool: """ Check if sender is blocked for this recipient Args: recipient: Recipient email address sender: Sender email address (may include name) worker_name: Worker name for logging Returns: True if sender is blocked """ patterns = self.dynamodb.get_blocked_patterns(recipient) if not patterns: return False sender_clean = parseaddr(sender)[1].lower() for pattern in patterns: if fnmatch.fnmatch(sender_clean, pattern.lower()): log( f"⛔ BLOCKED: Sender {sender_clean} matches pattern '{pattern}' " f"for inbox {recipient}", 'WARNING', worker_name ) return True return False def batch_check_blocked_senders( self, recipients: List[str], sender: str, worker_name: str ) -> Dict[str, bool]: """ Batch check if sender is blocked for multiple recipients (more efficient) Args: recipients: List of recipient email addresses sender: Sender email address worker_name: Worker name for logging Returns: Dictionary mapping recipient -> is_blocked (bool) """ # Get all blocked patterns in one batch call patterns_by_recipient = self.dynamodb.batch_get_blocked_patterns(recipients) sender_clean = parseaddr(sender)[1].lower() result = {} for recipient in recipients: patterns = patterns_by_recipient.get(recipient, []) is_blocked = False for pattern in patterns: if fnmatch.fnmatch(sender_clean, pattern.lower()): log( f"⛔ BLOCKED: Sender {sender_clean} matches pattern '{pattern}' " f"for inbox {recipient}", 'WARNING', worker_name ) is_blocked = True break result[recipient] = is_blocked return result