aknuth/email-amazon
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

cleanup

Browse Source
This commit is contained in:
Andreas Knuth
2026-01-16 22:16:09 -06:00
parent 5122082914
commit f7fe285200
7 changed files with 251 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

201
DMS/content_filter.py
View File

@@ -2,6 +2,8 @@
"""
Postfix Content Filter for Internal Email Processing
Handles forwarding and auto-reply for local deliveries
Version: 2.0 (Optimized)
"""
import os
@@ -25,27 +27,62 @@ logging.basicConfig(
)
# AWS Configuration
AWS_REGION = 'us-east-2'
DYNAMODB_RULES_TABLE = 'email-rules'
AWS_REGION = os.environ.get('AWS_REGION', 'us-east-2')
DYNAMODB_RULES_TABLE = os.environ.get('DYNAMODB_RULES_TABLE', 'email-rules')
# SMTP Configuration
REINJECT_HOST = 'localhost'
REINJECT_PORT = 10026
REINJECT_HOST = os.environ.get('REINJECT_HOST', 'localhost')
REINJECT_PORT = int(os.environ.get('REINJECT_PORT', '10026'))
# Cache for DynamoDB rules (TTL 5 minutes)
# Cache Configuration
CACHE_TTL_MINUTES = int(os.environ.get('CACHE_TTL_MINUTES', '5'))
CACHE_TTL = timedelta(minutes=CACHE_TTL_MINUTES)
# Auto-reply throttling (prevent sending multiple auto-replies to same sender)
# Key: (recipient, sender), Value: last_sent_timestamp
AUTOREPLY_SENT = {}
AUTOREPLY_THROTTLE = timedelta(hours=24) # Only one auto-reply per sender per day
# Cache for DynamoDB rules
RULES_CACHE = {}
CACHE_TTL = timedelta(minutes=5)
# Initialize boto3 (lazy import to catch errors)
DYNAMODB_AVAILABLE = False
try:
import boto3
dynamodb = boto3.resource('dynamodb', region_name=AWS_REGION)
rules_table = dynamodb.Table(DYNAMODB_RULES_TABLE)
# Test connection
rules_table.table_status
DYNAMODB_AVAILABLE = True
logging.info("DynamoDB connection initialized")
logging.info("✓ DynamoDB connection initialized")
except Exception as e:
DYNAMODB_AVAILABLE = False
logging.error(f"DynamoDB initialization failed: {e}")
logging.error(f"✗ DynamoDB initialization failed: {e}")
logging.warning("Auto-reply and forwarding will be DISABLED")
def extract_domain(email_addr):
"""
Extract domain from email address, handling various formats
Examples:
"user@example.com" -> "example.com"
"Name <user@example.com>" -> "example.com"
"invalid" -> ""
"""
if not email_addr or '@' not in email_addr:
return ''
# parseaddr handles "Name <email@domain.com>" format
_, addr = parseaddr(email_addr)
if '@' not in addr:
return ''
try:
domain = addr.split('@')[1].lower()
return domain
except (IndexError, AttributeError):
return ''
def get_email_rules(email_address):
"""Fetch forwarding and auto-reply rules from DynamoDB with caching"""
@@ -54,23 +91,40 @@ def get_email_rules(email_address):
now = datetime.now()
cache_key = email_address.lower()
if cache_key in RULES_CACHE and now - RULES_CACHE[cache_key]['time'] < CACHE_TTL:
logging.debug(f"Cache hit for {email_address}")
return RULES_CACHE[cache_key]['rules']
# Check cache
if cache_key in RULES_CACHE:
cached_entry = RULES_CACHE[cache_key]
if now - cached_entry['time'] < CACHE_TTL:
logging.debug(f"Cache hit for {email_address}")
return cached_entry['rules']
# Fetch from DynamoDB
try:
response = rules_table.get_item(Key={'email_address': email_address})
item = response.get('Item', {})
if item:
logging.info(f"Rules found for {email_address}: forwards={len(item.get('forwards', []))}, ooo={item.get('ooo_active', False)}")
forwards_count = len(item.get('forwards', []))
ooo_active = item.get('ooo_active', False)
logging.info(f"Rules for {email_address}: forwards={forwards_count}, ooo={ooo_active}")
else:
logging.debug(f"No rules found for {email_address}")
# Update cache
RULES_CACHE[cache_key] = {'rules': item, 'time': now}
return item
except Exception as e:
logging.error(f"DynamoDB error for {email_address}: {e}")
return {}
def should_send_autoreply(original_msg, sender_addr):
"""Check if we should send auto-reply to this sender"""
def should_send_autoreply(original_msg, sender_addr, recipient_addr):
"""
Check if we should send auto-reply to this sender
Returns: (bool, str) - (should_send, reason_if_not)
"""
sender_lower = sender_addr.lower()
# Don't reply to automated senders
@@ -80,24 +134,43 @@ def should_send_autoreply(original_msg, sender_addr):
'noreply',
'no-reply',
'donotreply',
'do-not-reply',
'bounce',
'amazonses.com'
'amazonses.com',
'notification',
]
for pattern in blocked_patterns:
if pattern in sender_lower:
logging.info(f"Skipping auto-reply to automated sender: {sender_addr}")
return False
return (False, f"automated sender pattern: {pattern}")
# Check for auto-submitted header to prevent loops (RFC 3834)
if original_msg.get('Auto-Submitted', '').startswith('auto-'):
logging.info(f"Skipping auto-reply due to Auto-Submitted header: {sender_addr}")
return False
auto_submitted = original_msg.get('Auto-Submitted', '')
if auto_submitted and auto_submitted.lower().startswith('auto-'):
return (False, f"Auto-Submitted header: {auto_submitted}")
return True
# Check precedence header (mailing lists, bulk mail)
precedence = original_msg.get('Precedence', '').lower()
if precedence in ['bulk', 'list', 'junk']:
return (False, f"Precedence: {precedence}")
# Check List-* headers (mailing lists)
if original_msg.get('List-Id') or original_msg.get('List-Unsubscribe'):
return (False, "mailing list headers detected")
# Throttle: Only send one auto-reply per sender per 24 hours
throttle_key = (recipient_addr.lower(), sender_lower)
if throttle_key in AUTOREPLY_SENT:
last_sent = AUTOREPLY_SENT[throttle_key]
if datetime.now() - last_sent < AUTOREPLY_THROTTLE:
time_left = AUTOREPLY_THROTTLE - (datetime.now() - last_sent)
hours_left = int(time_left.total_seconds() / 3600)
return (False, f"throttled (sent {hours_left}h ago)")
return (True, "")
def send_autoreply(original_msg, recipient_rules, recipient_addr):
"""Send auto-reply if enabled"""
"""Send auto-reply if enabled and appropriate"""
if not recipient_rules.get('ooo_active'):
return
@@ -109,7 +182,14 @@ def send_autoreply(original_msg, recipient_rules, recipient_addr):
# Extract email from "Name <email>" format
sender_name, sender_addr = parseaddr(sender)
if not should_send_autoreply(original_msg, sender_addr):
if not sender_addr or '@' not in sender_addr:
logging.warning(f"Invalid sender address: {sender}, skipping auto-reply")
return
# Check if we should send auto-reply
should_send, reason = should_send_autoreply(original_msg, sender_addr, recipient_addr)
if not should_send:
logging.info(f"Skipping auto-reply to {sender_addr}: {reason}")
return
subject = original_msg.get('Subject', 'No Subject')
@@ -136,6 +216,7 @@ def send_autoreply(original_msg, recipient_rules, recipient_addr):
reply['Message-ID'] = make_msgid()
reply['Auto-Submitted'] = 'auto-replied' # RFC 3834
reply['Precedence'] = 'bulk'
reply['X-Auto-Response-Suppress'] = 'All' # Microsoft Exchange
if message_id:
reply['In-Reply-To'] = message_id
@@ -145,11 +226,16 @@ def send_autoreply(original_msg, recipient_rules, recipient_addr):
try:
with smtplib.SMTP(REINJECT_HOST, REINJECT_PORT, timeout=30) as smtp:
smtp.send_message(reply)
# Update throttle timestamp
throttle_key = (recipient_addr.lower(), sender_addr.lower())
AUTOREPLY_SENT[throttle_key] = datetime.now()
logging.info(f"✓ Sent auto-reply from {recipient_addr} to {sender_addr}")
except Exception as e:
logging.error(f"✗ Auto-reply failed: {e}")
def send_forwards(original_msg_bytes, recipient_rules, recipient_addr):
def send_forwards(original_msg_bytes, recipient_rules, recipient_addr, sender_addr):
"""Forward email to configured addresses"""
forwards = recipient_rules.get('forwards', [])
if not forwards:
@@ -157,17 +243,27 @@ def send_forwards(original_msg_bytes, recipient_rules, recipient_addr):
for forward_addr in forwards:
try:
# Validate forward address
if '@' not in forward_addr:
logging.warning(f"Invalid forward address: {forward_addr}, skipping")
continue
# Parse message again for clean forwarding
msg = message_from_binary_file(BytesIO(original_msg_bytes))
# Add forwarding headers
msg['X-Forwarded-For'] = recipient_addr
msg['X-Original-To'] = recipient_addr
msg['X-Forwarded-By'] = 'content_filter.py'
# Preserve original sender in envelope
# (This way replies go to original sender, not to recipient)
envelope_sender = sender_addr if sender_addr else recipient_addr
# Send via local SMTP
with smtplib.SMTP(REINJECT_HOST, REINJECT_PORT, timeout=30) as smtp:
smtp.sendmail(
from_addr=recipient_addr,
from_addr=envelope_sender,
to_addrs=[forward_addr],
msg=msg.as_bytes()
)
@@ -175,6 +271,20 @@ def send_forwards(original_msg_bytes, recipient_rules, recipient_addr):
except Exception as e:
logging.error(f"✗ Forward to {forward_addr} failed: {e}")
def is_internal_mail(sender, recipient):
"""
Check if this is internal mail (same domain)
This is a safety check in addition to transport_maps filtering
"""
sender_domain = extract_domain(sender)
recipient_domain = extract_domain(recipient)
if not sender_domain or not recipient_domain:
return False
return sender_domain == recipient_domain
def main():
"""Main content filter logic"""
if len(sys.argv) < 3:
@@ -184,53 +294,74 @@ def main():
sender = sys.argv[1]
recipients = sys.argv[2:]
logging.info(f"Processing email from {sender} to {', '.join(recipient)}")
logging.info(f"Processing email from {sender} to {', '.join(recipients)}")
# Read email from stdin
try:
msg_bytes = sys.stdin.buffer.read()
if not msg_bytes:
logging.error("No email data received on stdin")
sys.exit(75) # EX_TEMPFAIL
msg = message_from_binary_file(BytesIO(msg_bytes))
except Exception as e:
logging.error(f"Failed to read email: {e}")
sys.exit(75) # EX_TEMPFAIL
# Process each recipient
processed_count = 0
for recipient in recipients:
try:
# Check if mail is internal (same domain) - skip if external
_, recipient_domain = parseaddr(recipient)[1].rsplit('@', 1) if '@' in recipient else ('', '')
_, sender_domain = parseaddr(sender)[1].rsplit('@', 1) if '@' in sender else ('', '')
if recipient_domain != sender_domain:
logging.info(f"Skipping external mail to {recipient} (sender domain: {sender_domain})")
# Safety check: Only process internal mail
# (transport_maps should already filter, but defense-in-depth)
if not is_internal_mail(sender, recipient):
logging.debug(f"Skipping external mail: {sender} -> {recipient}")
continue
# Fetch rules from DynamoDB
rules = get_email_rules(recipient)
if rules:
processed_count += 1
# Send auto-reply if configured
send_autoreply(msg, rules, recipient)
if rules.get('ooo_active'):
send_autoreply(msg, rules, recipient)
# Send forwards if configured
send_forwards(msg_bytes, rules, recipient)
if rules.get('forwards'):
send_forwards(msg_bytes, rules, recipient, sender)
else:
logging.debug(f"No rules for {recipient}")
except Exception as e:
logging.error(f"Error processing rules for {recipient}: {e}")
import traceback
logging.error(traceback.format_exc())
if processed_count > 0:
logging.info(f"Processed rules for {processed_count}/{len(recipients)} recipients")
# Re-inject original email for normal delivery
try:
with smtplib.SMTP(REINJECT_HOST, REINJECT_PORT, timeout=30) as smtp:
smtp.sendmail(sender, recipients, msg_bytes)
logging.info(f"✓ Delivered to {', '.join(recipient)}")
logging.info(f"✓ Delivered to {', '.join(recipients)}")
sys.exit(0)
except Exception as e:
logging.error(f"✗ Delivery failed: {e}")
import traceback
logging.error(traceback.format_exc())
sys.exit(75) # EX_TEMPFAIL - Postfix will retry
if __name__ == '__main__':
try:
main()
except KeyboardInterrupt:
logging.info("Interrupted by user")
sys.exit(1)
except Exception as e:
logging.error(f"Fatal error: {e}")
import traceback
logging.error(traceback.format_exc())
sys.exit(75)