diff --git a/lambda/relay-outbound.py b/lambda/relay-outbound.py new file mode 100644 index 0000000..349ce31 --- /dev/null +++ b/lambda/relay-outbound.py @@ -0,0 +1,284 @@ +""" +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): + """ + 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 + + 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, + ) + + +# ------------------------------------------------------------ +# 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) + 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}