377 lines
14 KiB
Python
377 lines
14 KiB
Python
import sys
|
|
from flask import Flask, request, jsonify, g
|
|
import smtplib
|
|
import base64
|
|
import gzip
|
|
import logging
|
|
import os
|
|
import time
|
|
import boto3
|
|
from pathlib import Path
|
|
from dotenv import load_dotenv
|
|
from email.parser import BytesParser
|
|
from email.policy import default
|
|
from email.utils import getaddresses
|
|
|
|
# Python-Version prüfen
|
|
if sys.version_info < (3, 12):
|
|
raise RuntimeError("Python 3.12 oder höher erforderlich")
|
|
|
|
# .env-Datei laden
|
|
load_dotenv()
|
|
|
|
app = Flask(__name__)
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Konfiguration
|
|
SMTP_HOST = "localhost" # Host-Netzwerkmodus
|
|
SMTP_PORT = 25 # Fest auf Port 25 ohne TLS
|
|
API_TOKEN = os.environ.get('API_TOKEN')
|
|
MAIL_DIR = "/var/mail" # Standard Maildir-Pfad
|
|
AWS_REGION = os.environ.get('AWS_REGION', 'us-east-1')
|
|
|
|
if not API_TOKEN:
|
|
raise ValueError("API_TOKEN Umgebungsvariable nicht gesetzt")
|
|
|
|
# S3-Client initialisieren
|
|
s3_client = boto3.client('s3', region_name=AWS_REGION)
|
|
|
|
# Cache für verarbeitete Requests
|
|
processed_requests = {}
|
|
|
|
logger.info(f"API_TOKEN loaded: {API_TOKEN}")
|
|
|
|
def load_domains_config():
|
|
"""Lädt die Domain-Konfiguration"""
|
|
# Hier könntest du eine echte Konfigurationsdatei laden
|
|
# Für jetzt als Beispiel hardcoded
|
|
return {
|
|
"andreasknuth.de": {"bucket": "andreasknuth-de-emails"},
|
|
# Weitere Domains können hier hinzugefügt werden
|
|
}
|
|
|
|
def get_bucket_name_for_domain(domain):
|
|
"""Konvertiert Domain-Namen zu S3-Bucket-Namen"""
|
|
return domain.replace(".", "-") + "-emails"
|
|
|
|
def mark_email_as_processed(bucket_name, key):
|
|
"""Markiert eine E-Mail als erfolgreich verarbeitet"""
|
|
try:
|
|
# Füge Metadata hinzu, um zu markieren, dass die E-Mail verarbeitet wurde
|
|
s3_client.copy_object(
|
|
Bucket=bucket_name,
|
|
Key=key,
|
|
CopySource={'Bucket': bucket_name, 'Key': key},
|
|
Metadata={
|
|
'processed': 'true',
|
|
'processed_timestamp': str(int(time.time())),
|
|
'processor': 'rest-api'
|
|
},
|
|
MetadataDirective='REPLACE'
|
|
)
|
|
logger.info(f"E-Mail {key} als verarbeitet markiert")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Fehler beim Markieren der E-Mail {key}: {str(e)}")
|
|
return False
|
|
|
|
def get_unprocessed_emails(bucket_name):
|
|
"""Holt alle unverarbeiteten E-Mails aus einem S3-Bucket"""
|
|
try:
|
|
unprocessed_emails = []
|
|
paginator = s3_client.get_paginator('list_objects_v2')
|
|
|
|
for page in paginator.paginate(Bucket=bucket_name):
|
|
if 'Contents' not in page:
|
|
continue
|
|
|
|
for obj in page['Contents']:
|
|
key = obj['Key']
|
|
|
|
# Prüfe Metadata der E-Mail
|
|
try:
|
|
head_response = s3_client.head_object(Bucket=bucket_name, Key=key)
|
|
metadata = head_response.get('Metadata', {})
|
|
|
|
# Wenn nicht als verarbeitet markiert, zur Liste hinzufügen
|
|
if metadata.get('processed') != 'true':
|
|
unprocessed_emails.append({
|
|
'key': key,
|
|
'size': obj['Size'],
|
|
'last_modified': obj['LastModified'].isoformat(),
|
|
'metadata': metadata
|
|
})
|
|
except Exception as e:
|
|
logger.warning(f"Fehler beim Prüfen der Metadata für {key}: {str(e)}")
|
|
# Wenn Metadata nicht gelesen werden kann, als unverarbeitet betrachten
|
|
unprocessed_emails.append({
|
|
'key': key,
|
|
'size': obj['Size'],
|
|
'last_modified': obj['LastModified'].isoformat(),
|
|
'metadata': {}
|
|
})
|
|
|
|
return unprocessed_emails
|
|
except Exception as e:
|
|
logger.error(f"Fehler beim Abrufen unverarbeiteter E-Mails: {str(e)}")
|
|
return []
|
|
|
|
@app.route('/health', methods=['GET'])
|
|
def health_check():
|
|
"""Erweiterte Gesundheitsprüfung"""
|
|
|
|
# Basis-Gesundheitscheck
|
|
health_status = {
|
|
"status": "OK",
|
|
"message": "S3 E-Mail Processor API ist aktiv",
|
|
"timestamp": int(time.time()),
|
|
"version": "2.0",
|
|
"request_id": getattr(g, 'request_id', 'health-check')
|
|
}
|
|
|
|
# Erweiterte Checks
|
|
try:
|
|
# Maildir-Zugriff prüfen
|
|
mail_path = Path(MAIL_DIR)
|
|
if mail_path.exists() and mail_path.is_dir():
|
|
health_status["mail_dir"] = "accessible"
|
|
else:
|
|
health_status["mail_dir"] = "not_accessible"
|
|
health_status["status"] = "WARNING"
|
|
|
|
# Domain-Konfiguration prüfen
|
|
domains = load_domains_config()
|
|
health_status["configured_domains"] = len(domains)
|
|
health_status["domains"] = list(domains.keys())
|
|
|
|
# Cache-Status
|
|
health_status["request_cache_size"] = len(processed_requests)
|
|
|
|
# SMTP-Verbindung testen
|
|
try:
|
|
with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=5) as smtp:
|
|
smtp.noop()
|
|
health_status["smtp_connection"] = "OK"
|
|
except Exception as e:
|
|
health_status["smtp_connection"] = f"ERROR: {str(e)}"
|
|
health_status["status"] = "WARNING"
|
|
|
|
# S3-Verbindung testen
|
|
try:
|
|
s3_client.list_buckets()
|
|
health_status["s3_connection"] = "OK"
|
|
except Exception as e:
|
|
health_status["s3_connection"] = f"ERROR: {str(e)}"
|
|
health_status["status"] = "WARNING"
|
|
|
|
except Exception as e:
|
|
health_status["status"] = "ERROR"
|
|
health_status["error"] = str(e)
|
|
|
|
return jsonify(health_status)
|
|
|
|
@app.route('/retry/<domain>', methods=['GET'])
|
|
def retry_domain_emails(domain):
|
|
"""Verarbeitet alle unverarbeiteten E-Mails für eine Domain"""
|
|
auth_header = request.headers.get('Authorization')
|
|
if not auth_header or auth_header != f'Bearer {API_TOKEN}':
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
|
|
request_id = f"retry-{domain}-{int(time.time())}"
|
|
bucket_name = get_bucket_name_for_domain(domain)
|
|
|
|
logger.info(f"[{request_id}] Retry-Verarbeitung für Domain: {domain}, Bucket: {bucket_name}")
|
|
|
|
try:
|
|
# Hole alle unverarbeiteten E-Mails
|
|
unprocessed_emails = get_unprocessed_emails(bucket_name)
|
|
|
|
if not unprocessed_emails:
|
|
return jsonify({
|
|
'message': f'Keine unverarbeiteten E-Mails für Domain {domain} gefunden',
|
|
'domain': domain,
|
|
'bucket': bucket_name,
|
|
'processed_count': 0,
|
|
'request_id': request_id
|
|
}), 200
|
|
|
|
processed_count = 0
|
|
failed_count = 0
|
|
errors = []
|
|
|
|
for email_info in unprocessed_emails:
|
|
key = email_info['key']
|
|
|
|
try:
|
|
logger.info(f"[{request_id}] Verarbeite E-Mail: {key}")
|
|
|
|
# E-Mail aus S3 laden
|
|
response = s3_client.get_object(Bucket=bucket_name, Key=key)
|
|
email_content = response['Body'].read()
|
|
|
|
# E-Mail parsen
|
|
email_msg = BytesParser(policy=default).parsebytes(email_content)
|
|
|
|
# From-Adresse extrahieren
|
|
from_headers = email_msg.get_all('from', [])
|
|
if from_headers:
|
|
from_addr = getaddresses(from_headers)[0][1]
|
|
else:
|
|
from_addr = f'retry@{domain}'
|
|
|
|
# Empfänger extrahieren
|
|
to_addrs = [addr for _name, addr in getaddresses(email_msg.get_all('to', []))]
|
|
cc_addrs = [addr for _name, addr in getaddresses(email_msg.get_all('cc', []))]
|
|
bcc_addrs = [addr for _name, addr in getaddresses(email_msg.get_all('bcc', []))]
|
|
recipients = to_addrs + cc_addrs + bcc_addrs
|
|
|
|
if not recipients:
|
|
raise ValueError(f"Keine Empfänger in E-Mail {key} gefunden")
|
|
|
|
# An SMTP weiterleiten
|
|
with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as smtp:
|
|
smtp.mail(from_addr)
|
|
for recipient in recipients:
|
|
smtp.rcpt(recipient)
|
|
smtp.data(email_content)
|
|
|
|
# Als verarbeitet markieren
|
|
mark_email_as_processed(bucket_name, key)
|
|
processed_count += 1
|
|
|
|
logger.info(f"[{request_id}] E-Mail {key} erfolgreich verarbeitet")
|
|
|
|
except Exception as e:
|
|
failed_count += 1
|
|
error_msg = f"Fehler bei E-Mail {key}: {str(e)}"
|
|
errors.append(error_msg)
|
|
logger.error(f"[{request_id}] {error_msg}")
|
|
|
|
result = {
|
|
'message': f'Retry-Verarbeitung für Domain {domain} abgeschlossen',
|
|
'domain': domain,
|
|
'bucket': bucket_name,
|
|
'total_found': len(unprocessed_emails),
|
|
'processed_count': processed_count,
|
|
'failed_count': failed_count,
|
|
'request_id': request_id
|
|
}
|
|
|
|
if errors:
|
|
result['errors'] = errors
|
|
|
|
return jsonify(result), 200
|
|
|
|
except Exception as e:
|
|
logger.error(f"[{request_id}] Fehler bei Retry-Verarbeitung: {str(e)}")
|
|
return jsonify({
|
|
'error': str(e),
|
|
'domain': domain,
|
|
'bucket': bucket_name,
|
|
'request_id': request_id
|
|
}), 500
|
|
|
|
@app.route('/process/<domain>', methods=['POST'])
|
|
def process_email(domain):
|
|
"""Verarbeitet eine einzelne E-Mail von der Lambda-Funktion"""
|
|
auth_header = request.headers.get('Authorization')
|
|
if not auth_header or auth_header != f'Bearer {API_TOKEN}':
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
|
|
data = request.get_json()
|
|
if not data:
|
|
return jsonify({'error': 'Invalid JSON payload'}), 400
|
|
|
|
request_id = data.get('request_id')
|
|
email_content = data.get('email_content')
|
|
compressed = data.get('compressed', False)
|
|
s3_key = data.get('s3_key') # S3-Key für Marker-Funktionalität
|
|
s3_bucket = data.get('s3_bucket') # S3-Bucket für Marker-Funktionalität
|
|
|
|
logger.info(f"[{request_id}] Processing email for domain: {domain}")
|
|
if s3_key:
|
|
logger.info(f"[{request_id}] S3 Location: {s3_bucket}/{s3_key}")
|
|
|
|
try:
|
|
# Entkomprimieren, falls komprimiert
|
|
if compressed:
|
|
email_bytes = base64.b64decode(email_content)
|
|
email_content = gzip.decompress(email_bytes)
|
|
else:
|
|
email_content = base64.b64decode(email_content)
|
|
|
|
# E-Mail-Header parsen
|
|
email_msg = BytesParser(policy=default).parsebytes(email_content)
|
|
|
|
# From-Adresse extrahieren (konsistent mit Lambda)
|
|
from_headers = email_msg.get_all('from', [])
|
|
if from_headers:
|
|
from_addr = getaddresses(from_headers)[0][1]
|
|
else:
|
|
from_addr = f'lambda@{domain}' # Fallback mit Domain
|
|
|
|
# Empfänger aus allen relevanten Headern extrahieren (konsistent mit Lambda)
|
|
to_addrs = [addr for _name, addr in getaddresses(email_msg.get_all('to', []))]
|
|
cc_addrs = [addr for _name, addr in getaddresses(email_msg.get_all('cc', []))]
|
|
bcc_addrs = [addr for _name, addr in getaddresses(email_msg.get_all('bcc', []))]
|
|
recipients = to_addrs + cc_addrs + bcc_addrs
|
|
|
|
# Debug-Ausgabe für Empfänger
|
|
logger.info(f"[{request_id}] To-Adressen: {to_addrs}")
|
|
logger.info(f"[{request_id}] CC-Adressen: {cc_addrs}")
|
|
logger.info(f"[{request_id}] BCC-Adressen: {bcc_addrs}")
|
|
logger.info(f"[{request_id}] Alle Empfänger: {recipients}")
|
|
|
|
if not recipients:
|
|
# Zusätzliche Debug-Info
|
|
logger.error(f"[{request_id}] Verfügbare Header: {list(email_msg.keys())}")
|
|
logger.error(f"[{request_id}] To-Header roh: {email_msg.get('To')}")
|
|
logger.error(f"[{request_id}] CC-Header roh: {email_msg.get('Cc')}")
|
|
raise ValueError("Keine Empfänger (To/CC/BCC) in der E-Mail gefunden")
|
|
|
|
logger.info(f"[{request_id}] From: {from_addr}, Recipients: {recipients}")
|
|
|
|
# An Postfix weiterleiten - für jeden Empfänger einzeln
|
|
with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as smtp:
|
|
smtp.mail(from_addr)
|
|
for recipient in recipients:
|
|
smtp.rcpt(recipient)
|
|
smtp.data(email_content)
|
|
|
|
logger.info(f"[{request_id}] Email forwarded to Postfix for {domain} - {len(recipients)} recipients")
|
|
|
|
# E-Mail als verarbeitet markieren, falls S3-Informationen vorhanden
|
|
marked_as_processed = False
|
|
if s3_key and s3_bucket:
|
|
if mark_email_as_processed(s3_bucket, s3_key):
|
|
logger.info(f"[{request_id}] E-Mail {s3_key} als verarbeitet markiert")
|
|
marked_as_processed = True
|
|
else:
|
|
logger.warning(f"[{request_id}] Konnte E-Mail {s3_key} nicht als verarbeitet markieren")
|
|
elif s3_key:
|
|
# Fallback: Bucket-Name aus Domain ableiten
|
|
bucket_name = get_bucket_name_for_domain(domain)
|
|
if mark_email_as_processed(bucket_name, s3_key):
|
|
logger.info(f"[{request_id}] E-Mail {s3_key} als verarbeitet markiert (abgeleiteter Bucket: {bucket_name})")
|
|
marked_as_processed = True
|
|
else:
|
|
logger.warning(f"[{request_id}] Konnte E-Mail {s3_key} nicht als verarbeitet markieren")
|
|
|
|
return jsonify({
|
|
'message': 'Email processed',
|
|
'request_id': request_id,
|
|
'domain': domain,
|
|
'recipients_count': len(recipients),
|
|
'recipients': recipients,
|
|
'marked_as_processed': marked_as_processed,
|
|
's3_bucket': s3_bucket,
|
|
's3_key': s3_key
|
|
}), 200
|
|
|
|
except Exception as e:
|
|
logger.error(f"[{request_id}] Error processing email: {str(e)}")
|
|
return jsonify({'error': str(e), 'request_id': request_id}), 500
|
|
|
|
if __name__ == '__main__':
|
|
app.run(host='0.0.0.0', port=5000) |