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

avoid loops

This commit is contained in:
Andreas Knuth 2026-01-13 21:47:54 -06:00
parent 397a2f7d98
commit 7f9ac1c9e6
1 changed files with 26 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
unified-worker/unified_worker.py
View File

@ -468,7 +468,8 @@ def create_ooo_reply(original_parsed, recipient: str, ooo_msg: str, content_type
msg['Message-ID'] = make_msgid(domain=recipient.split('@')[1]) msg['Message-ID'] = make_msgid(domain=recipient.split('@')[1])
msg['In-Reply-To'] = original_parsed.get('Message-ID', '') msg['In-Reply-To'] = original_parsed.get('Message-ID', '')
msg['References'] = original_parsed.get('Message-ID', '') msg['References'] = original_parsed.get('Message-ID', '')
msg['Auto-Submitted'] = 'auto-replied' # Verhindert Loops msg['Auto-Submitted'] = 'auto-replied' # Standard header für Auto-Replies
msg['X-SES-Worker-Processed'] = 'ooo-reply' # Unser Loop-Prevention Header
# Body-Teil erstellen # Body-Teil erstellen
body_part = MIMEMultipart('alternative') body_part = MIMEMultipart('alternative')
@ -515,6 +516,7 @@ def create_forward_message(original_parsed, recipient: str, forward_to: str, ori
msg['Date'] = formatdate(localtime=True) msg['Date'] = formatdate(localtime=True)
msg['Message-ID'] = make_msgid(domain=recipient.split('@')[1]) msg['Message-ID'] = make_msgid(domain=recipient.split('@')[1])
msg['Reply-To'] = original_from msg['Reply-To'] = original_from
msg['X-SES-Worker-Processed'] = 'forwarded' # Unser Loop-Prevention Header
# Forward-Header als Text # Forward-Header als Text
text_body, html_body = extract_body_parts(original_parsed) text_body, html_body = extract_body_parts(original_parsed)
@ -885,6 +887,25 @@ def process_message(domain: str, message: dict, receive_count: int) -> bool:
try: try:
response = s3.get_object(Bucket=bucket, Key=key) response = s3.get_object(Bucket=bucket, Key=key)
raw_bytes = response['Body'].read() raw_bytes = response['Body'].read()
# LOOP DETECTION: Check if this message was already processed by our worker
# We use a unique header that only our worker sets
temp_parsed = BytesParser(policy=SMTPPolicy).parsebytes(raw_bytes)
# Check for OUR unique loop prevention header
x_worker_processed = temp_parsed.get('X-SES-Worker-Processed', '')
auto_submitted = temp_parsed.get('Auto-Submitted', '')
# Only skip if OUR header is present (not generic headers that others might set)
is_processed_by_us = bool(x_worker_processed)
is_our_auto_reply = auto_submitted == 'auto-replied' and x_worker_processed
if is_processed_by_us:
log(f"🔄 Loop prevention: Already processed by worker", 'INFO', worker_name)
skip_rules = True
else:
skip_rules = False
except s3.exceptions.NoSuchKey: except s3.exceptions.NoSuchKey:
if receive_count < 5: if receive_count < 5:
log(f"⏳ S3 Object not found yet (Attempt {receive_count}). Retrying...", 'WARNING', worker_name) log(f"⏳ S3 Object not found yet (Attempt {receive_count}). Retrying...", 'WARNING', worker_name)
@ -929,9 +950,10 @@ def process_message(domain: str, message: dict, receive_count: int) -> bool:
log(f"⚠ Parsing/Logic Error: {e}. Sending original.", 'WARNING', worker_name) log(f"⚠ Parsing/Logic Error: {e}. Sending original.", 'WARNING', worker_name)
from_addr_final = from_addr from_addr_final = from_addr
is_bounce = False is_bounce = False
skip_rules = False # Default if parsing failed
# 5. OOO & FORWARD LOGIC (vor SMTP-Versand, nicht für Bounces) # 5. OOO & FORWARD LOGIC (vor SMTP-Versand, nicht für Bounces oder bereits weitergeleitete)
if not is_bounce: if not is_bounce and not skip_rules:
for recipient in recipients: for recipient in recipients:
process_rules_for_recipient(recipient, parsed, domain, worker_name) process_rules_for_recipient(recipient, parsed, domain, worker_name)
@ -1282,4 +1304,4 @@ def main():
worker.start() worker.start()
if __name__ == '__main__': if __name__ == '__main__':
main() main()