318 lines
11 KiB
Python
318 lines
11 KiB
Python
"""
|
|
relay-outbound Lambda
|
|
|
|
Handles SES Configuration Set events forwarded via EventBridge.
|
|
Writes to two DynamoDB tables:
|
|
|
|
- ses-outbound-messages : EXISTING table for bounce forensics
|
|
(kept unchanged so existing tooling keeps working)
|
|
- ses-events : NEW table for billing-style aggregation
|
|
of sends/bounces/complaints
|
|
|
|
We list mailadmin domains to filter out events from non-mailadmin
|
|
addresses (e.g. SES console test sends).
|
|
"""
|
|
import json
|
|
import os
|
|
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import boto3
|
|
|
|
dynamodb = boto3.resource('dynamodb')
|
|
ses_client = boto3.client('ses')
|
|
|
|
# Existing bounce-detail table — preserved as-is
|
|
bounce_table = dynamodb.Table('ses-outbound-messages')
|
|
|
|
# New unified events table for billing
|
|
events_table = dynamodb.Table('ses-events')
|
|
|
|
# ------------------------------------------------------------
|
|
# Domain filter
|
|
# ------------------------------------------------------------
|
|
# We only want to track events for mailboxes managed by mailadmin.
|
|
# Easiest source-of-truth: SES verified domain identities. Any address
|
|
# whose domain is verified in SES is "ours". Cached for 5 minutes so
|
|
# we don't hammer SES on every event.
|
|
# ------------------------------------------------------------
|
|
|
|
_domain_cache = {'expires_at': 0, 'domains': set()}
|
|
_DOMAIN_CACHE_TTL_SEC = 300
|
|
|
|
|
|
def get_managed_domains():
|
|
"""Return the set of SES verified domain identities (lowercased)."""
|
|
now = time.time()
|
|
if _domain_cache['expires_at'] > now and _domain_cache['domains']:
|
|
return _domain_cache['domains']
|
|
|
|
try:
|
|
# ListIdentities is paginated; for ~50 domains one page is enough,
|
|
# but handle pagination defensively.
|
|
domains = set()
|
|
next_token = None
|
|
while True:
|
|
kwargs = {'IdentityType': 'Domain', 'MaxItems': 100}
|
|
if next_token:
|
|
kwargs['NextToken'] = next_token
|
|
resp = ses_client.list_identities(**kwargs)
|
|
for d in resp.get('Identities', []):
|
|
domains.add(d.lower())
|
|
next_token = resp.get('NextToken')
|
|
if not next_token:
|
|
break
|
|
|
|
_domain_cache['domains'] = domains
|
|
_domain_cache['expires_at'] = now + _DOMAIN_CACHE_TTL_SEC
|
|
return domains
|
|
except Exception as exc:
|
|
print(f"WARN: could not list SES identities, falling back to cached set: {exc}")
|
|
return _domain_cache['domains']
|
|
|
|
|
|
def domain_of(email):
|
|
if not email or '@' not in email:
|
|
return None
|
|
return email.split('@', 1)[1].strip().lower()
|
|
|
|
|
|
def is_managed_address(email):
|
|
d = domain_of(email)
|
|
if not d:
|
|
return False
|
|
return d in get_managed_domains()
|
|
|
|
|
|
# ------------------------------------------------------------
|
|
# DynamoDB helpers
|
|
# ------------------------------------------------------------
|
|
|
|
def _ttl_two_years_from_now():
|
|
return int((datetime.now(timezone.utc) + timedelta(days=730)).timestamp())
|
|
|
|
|
|
def _month_bucket(ts_iso):
|
|
"""
|
|
Convert a SES event timestamp (ISO 8601) to 'YYYY-MM'.
|
|
Falls back to current UTC month if parsing fails.
|
|
"""
|
|
try:
|
|
if ts_iso.endswith('Z'):
|
|
ts_iso = ts_iso[:-1] + '+00:00'
|
|
dt = datetime.fromisoformat(ts_iso)
|
|
except Exception:
|
|
dt = datetime.now(timezone.utc)
|
|
return dt.strftime('%Y-%m')
|
|
|
|
|
|
def write_event_row(*, event_type, source, recipients, message_id, timestamp_iso,
|
|
size_bytes=None, subject=None, bounce_type=None, complaint_type=None,
|
|
smtp_response=None, reporting_mta=None):
|
|
"""
|
|
Write a row to the unified ses-events table.
|
|
|
|
Skips silently if the source address isn't from a managed domain
|
|
(per user choice: ignore non-mailadmin events).
|
|
"""
|
|
if not is_managed_address(source):
|
|
print(f"Skipping event for non-managed sender: {source}")
|
|
return
|
|
|
|
domain = domain_of(source)
|
|
ym = _month_bucket(timestamp_iso)
|
|
|
|
pk = f"{domain}#{source}#{ym}"
|
|
sk = f"{timestamp_iso}#{message_id}"
|
|
|
|
item = {
|
|
'pk': pk,
|
|
'sk': sk,
|
|
'event_type': event_type,
|
|
'from': source,
|
|
'domain': domain,
|
|
'recipients': recipients or [],
|
|
'message_id': message_id or '',
|
|
'timestamp': timestamp_iso,
|
|
'ttl_unix': _ttl_two_years_from_now(),
|
|
}
|
|
if size_bytes is not None:
|
|
item['size_bytes'] = int(size_bytes)
|
|
if subject:
|
|
item['subject'] = subject[:200]
|
|
if bounce_type:
|
|
item['bounce_type'] = bounce_type
|
|
if complaint_type:
|
|
item['complaint_type'] = complaint_type
|
|
if smtp_response:
|
|
item['smtp_response'] = smtp_response[:500]
|
|
if reporting_mta:
|
|
item['reporting_mta'] = reporting_mta
|
|
|
|
events_table.put_item(Item=item)
|
|
print(f"ses-events <- {event_type} {pk} {sk}")
|
|
|
|
|
|
# ------------------------------------------------------------
|
|
# Existing bounce handler (unchanged behavior, plus the new write)
|
|
# ------------------------------------------------------------
|
|
|
|
def handle_bounce(detail):
|
|
bounce = detail.get('bounce', {}) or {}
|
|
mail = detail.get('mail', {}) or {}
|
|
|
|
feedback_id = bounce.get('feedbackId')
|
|
bounce_type = bounce.get('bounceType', 'Unknown')
|
|
bounce_subtype = bounce.get('bounceSubType', 'Unknown')
|
|
bounced_recipients = [r.get('emailAddress') for r in bounce.get('bouncedRecipients', []) if r.get('emailAddress')]
|
|
original_source = mail.get('source')
|
|
original_message_id = mail.get('messageId')
|
|
timestamp = bounce.get('timestamp') or mail.get('timestamp') or datetime.now(timezone.utc).isoformat()
|
|
|
|
print(f"Bounce: id={feedback_id} from={original_source} type={bounce_type}/{bounce_subtype}")
|
|
|
|
# 1) EXISTING: store full bounce forensics in ses-outbound-messages
|
|
bounce_table.put_item(Item={
|
|
'MessageId': feedback_id,
|
|
'original_message_id': original_message_id,
|
|
'original_source': original_source,
|
|
'bounceType': bounce_type,
|
|
'bounceSubType': bounce_subtype,
|
|
'bouncedRecipients': bounced_recipients,
|
|
'timestamp': timestamp,
|
|
'event_type': 'bounce',
|
|
})
|
|
|
|
# 2) NEW: also record into ses-events for billing aggregation
|
|
write_event_row(
|
|
event_type='bounce',
|
|
source=original_source,
|
|
recipients=bounced_recipients,
|
|
message_id=original_message_id or feedback_id,
|
|
timestamp_iso=timestamp,
|
|
bounce_type=f"{bounce_type}/{bounce_subtype}",
|
|
)
|
|
|
|
|
|
# ------------------------------------------------------------
|
|
# New: send handler
|
|
# ------------------------------------------------------------
|
|
|
|
def handle_send(detail):
|
|
mail = detail.get('mail', {}) or {}
|
|
source = mail.get('source')
|
|
message_id = mail.get('messageId')
|
|
timestamp = mail.get('timestamp') or datetime.now(timezone.utc).isoformat()
|
|
|
|
# SES events do not include the message size directly. The closest
|
|
# info available is mail.headersTruncated and mail.commonHeaders, but
|
|
# neither contains the byte size. We approximate from the sum of all
|
|
# known header values + 0 (we'll log 0 if unknown). If size_bytes is
|
|
# critical, the alternative is to request the 'sendingPoolMetadata'
|
|
# via SES API, but for billing-grade tracking the SES Send Quota
|
|
# itself is per-message anyway.
|
|
common = mail.get('commonHeaders', {}) or {}
|
|
subject = common.get('subject')
|
|
|
|
# Recipients: prefer 'to' from commonHeaders, fall back to destination
|
|
to_field = common.get('to') or mail.get('destination') or []
|
|
if isinstance(to_field, str):
|
|
to_field = [to_field]
|
|
|
|
# Try to get size from SES if exposed (newer event format includes it
|
|
# under 'mail.headersTruncated' — not always present)
|
|
size_bytes = mail.get('messageSize') or mail.get('size') or 0
|
|
|
|
print(f"Send: id={message_id} from={source} to={to_field} size={size_bytes}")
|
|
|
|
write_event_row(
|
|
event_type='send',
|
|
source=source,
|
|
recipients=to_field,
|
|
message_id=message_id,
|
|
timestamp_iso=timestamp,
|
|
size_bytes=size_bytes,
|
|
subject=subject,
|
|
)
|
|
|
|
|
|
# ------------------------------------------------------------
|
|
# New: complaint handler
|
|
# ------------------------------------------------------------
|
|
|
|
def handle_complaint(detail):
|
|
complaint = detail.get('complaint', {}) or {}
|
|
mail = detail.get('mail', {}) or {}
|
|
|
|
source = mail.get('source')
|
|
message_id = mail.get('messageId')
|
|
timestamp = complaint.get('timestamp') or mail.get('timestamp') or datetime.now(timezone.utc).isoformat()
|
|
complaint_type = complaint.get('complaintFeedbackType') or 'unknown'
|
|
complained_recipients = [r.get('emailAddress') for r in complaint.get('complainedRecipients', []) if r.get('emailAddress')]
|
|
|
|
print(f"Complaint: id={message_id} from={source} type={complaint_type}")
|
|
|
|
write_event_row(
|
|
event_type='complaint',
|
|
source=source,
|
|
recipients=complained_recipients,
|
|
message_id=message_id,
|
|
timestamp_iso=timestamp,
|
|
complaint_type=complaint_type,
|
|
)
|
|
|
|
# ------------------------------------------------------------
|
|
# New: delivery handler
|
|
# ------------------------------------------------------------
|
|
def handle_delivery(detail):
|
|
delivery = detail.get('delivery', {}) or {}
|
|
mail = detail.get('mail', {}) or {}
|
|
|
|
source = mail.get('source')
|
|
message_id = mail.get('messageId')
|
|
timestamp = delivery.get('timestamp') or mail.get('timestamp') or datetime.now(timezone.utc).isoformat()
|
|
recipients = delivery.get('recipients') or []
|
|
smtp_response = delivery.get('smtpResponse') # wörtliche M365-Antwort
|
|
reporting_mta = delivery.get('reportingMTA')
|
|
processing_ms = delivery.get('processingTimeMillis')
|
|
|
|
print(f"Delivery: id={message_id} to={recipients} smtp={smtp_response!r} mta={reporting_mta}")
|
|
|
|
write_event_row(
|
|
event_type='delivery',
|
|
source=source,
|
|
recipients=recipients,
|
|
message_id=message_id,
|
|
timestamp_iso=timestamp,
|
|
smtp_response=smtp_response,
|
|
reporting_mta=reporting_mta,
|
|
)
|
|
|
|
# ------------------------------------------------------------
|
|
# Entry point
|
|
# ------------------------------------------------------------
|
|
|
|
def lambda_handler(event, context):
|
|
print(f"Received event: {json.dumps(event)[:500]}")
|
|
|
|
event_type = event.get('detail-type')
|
|
detail = event.get('detail', {}) or {}
|
|
|
|
try:
|
|
if event_type == 'Email Bounced':
|
|
handle_bounce(detail)
|
|
elif event_type == 'Email Sent':
|
|
handle_send(detail)
|
|
elif event_type == 'Email Complaint Received':
|
|
handle_complaint(detail)
|
|
elif event_type == 'Email Delivered':
|
|
handle_delivery(detail)
|
|
else:
|
|
print(f"Ignoring event type: {event_type}")
|
|
except Exception as exc:
|
|
# Don't fail the Lambda — losing one event is preferable to
|
|
# endless EventBridge retries that pile up DLQs.
|
|
print(f"ERROR processing {event_type}: {exc}")
|
|
|
|
return {'statusCode': 200}
|