387 lines
16 KiB
Python
387 lines
16 KiB
Python
import asyncio
|
|
import ipaddress
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
|
|
from aiosmtpd.smtp import SMTP
|
|
from sqlalchemy import func, select
|
|
|
|
from app.config import get_settings
|
|
from app.database import SessionFactory, mongo_db
|
|
from app.models import Account
|
|
from app.services.external_reply_service import log_original_external_message
|
|
from app.services.handoff_mirror_policy import (
|
|
GENERATED_BY_CUSTOMER,
|
|
MESSAGE_ORIGIN_EXTERNAL_CUSTOMER,
|
|
)
|
|
from app.services.human_handoff_service import MESSAGE_SCOPE_EXTERNAL
|
|
from app.services.human_handoff_service import (
|
|
maybe_trigger_handoff,
|
|
process_human_reply,
|
|
resolve_handoff_from_inbound,
|
|
start_human_initiated_handoff,
|
|
)
|
|
from app.services.message_ids import canonical_message_id, message_id_fields
|
|
from app.services.mime_parser import parse_mime
|
|
from app.services.reply_extractor import build_reply_display
|
|
from app.services.raw_mime_storage import delivery_digest, get_raw_mime_storage
|
|
from app.services.thread_resolver import participants_from_parsed, resolve_thread
|
|
from app.services.thread_service import ensure_thread_for_message, message_event_at
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def peer_is_allowed(session) -> bool:
|
|
if session is None or not getattr(session, "peer", None):
|
|
return True
|
|
try:
|
|
peer = ipaddress.ip_address(session.peer[0])
|
|
return any(
|
|
peer in ipaddress.ip_network(cidr)
|
|
for cidr in get_settings().inbound_allowed_cidrs
|
|
)
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def log_smtp_message_received(
|
|
*,
|
|
message_id: str | None,
|
|
in_reply_to: str | None,
|
|
references: list[str],
|
|
sender: str,
|
|
recipient: str,
|
|
subject: str,
|
|
resolved_scope: str,
|
|
resolved_thread_id,
|
|
resolution_method: str,
|
|
stored_message_id,
|
|
mailbox_id: str = "inbox",
|
|
) -> None:
|
|
logger.info(
|
|
"smtp_message_received message_id=%s in_reply_to=%s references=%s "
|
|
"from=%s to=%s subject=%s resolved_scope=%s resolved_thread_id=%s "
|
|
"resolution_method=%s stored_message_id=%s mailbox_id=%s",
|
|
message_id,
|
|
in_reply_to,
|
|
" ".join(references or []),
|
|
sender,
|
|
recipient,
|
|
subject,
|
|
resolved_scope,
|
|
resolved_thread_id,
|
|
resolution_method,
|
|
stored_message_id,
|
|
mailbox_id,
|
|
)
|
|
|
|
|
|
class InboundHandler:
|
|
async def handle_RCPT(self, server, session, envelope, address, rcpt_options):
|
|
if not peer_is_allowed(session):
|
|
logger.warning("Rejected inbound SMTP peer %s", session.peer[0])
|
|
return "550 5.7.1 Delivery from this host is not authorized"
|
|
normalized = address.strip().lower()
|
|
async with SessionFactory() as database_session:
|
|
account = await database_session.scalar(
|
|
select(Account).where(func.lower(Account.email) == normalized)
|
|
)
|
|
if account is None:
|
|
return "550 5.1.1 Recipient address rejected: user unknown"
|
|
envelope.rcpt_tos.append(normalized)
|
|
envelope.rcpt_options.extend(rcpt_options)
|
|
if not hasattr(envelope, "account_ids"):
|
|
envelope.account_ids = {}
|
|
envelope.account_ids[normalized] = str(account.id)
|
|
logger.info(
|
|
"smtp_rcpt_accepted mail_from=%s rcpt_to=%s account_id=%s",
|
|
envelope.mail_from.lower(),
|
|
normalized,
|
|
envelope.account_ids[normalized],
|
|
)
|
|
return "250 2.1.5 Recipient OK"
|
|
|
|
async def handle_DATA(self, server, session, envelope):
|
|
settings = get_settings()
|
|
raw = envelope.original_content
|
|
if len(raw) > settings.inbound_max_bytes:
|
|
return "552 5.3.4 Message size exceeds fixed maximum message size"
|
|
recipients = list(dict.fromkeys(address.lower() for address in envelope.rcpt_tos))
|
|
if not recipients:
|
|
return "554 5.5.1 No valid recipients"
|
|
|
|
delivery_key = delivery_digest(raw, recipients)
|
|
try:
|
|
tombstones = getattr(mongo_db, "message_tombstones", None)
|
|
if tombstones is not None and await tombstones.find_one(
|
|
{"delivery_key": delivery_key},
|
|
{"_id": 1},
|
|
):
|
|
return f"250 2.0.0 Message previously accepted as {delivery_key}"
|
|
|
|
logger.info(
|
|
"smtp_data_accepted mail_from=%s rcpt_to=%s delivery_key=%s bytes=%s",
|
|
envelope.mail_from.lower(),
|
|
",".join(recipients),
|
|
delivery_key,
|
|
len(raw),
|
|
)
|
|
|
|
parsed = parse_mime(raw)
|
|
body_text = parsed.pop("body_text")
|
|
body_html = parsed.pop("body_html")
|
|
stored = await get_raw_mime_storage().put(raw, delivery_key)
|
|
raw_mime = {
|
|
"bucket": stored.bucket,
|
|
"name": stored.name,
|
|
"generation": stored.generation,
|
|
"size": stored.size,
|
|
"crc32c": stored.crc32c,
|
|
}
|
|
now = datetime.now(timezone.utc)
|
|
sender_address = (parsed.get("sender") or {}).get("address", "").lower()
|
|
internet_message_id = canonical_message_id(parsed.get("message_id"))
|
|
id_fields = message_id_fields(internet_message_id)
|
|
|
|
external_recipients: list[str] = []
|
|
for recipient in recipients:
|
|
account_id = envelope.account_ids[recipient]
|
|
handoff = await resolve_handoff_from_inbound(
|
|
mongo_db,
|
|
account_id=account_id,
|
|
agent_email=recipient,
|
|
sender_address=sender_address,
|
|
in_reply_to=parsed.get("in_reply_to"),
|
|
references=parsed.get("references", []),
|
|
)
|
|
if handoff is None:
|
|
started = await start_human_initiated_handoff(
|
|
mongo_db,
|
|
account_id=account_id,
|
|
agent_email=recipient,
|
|
parsed=parsed,
|
|
body_text=body_text,
|
|
body_html=body_html,
|
|
raw_mime=raw_mime,
|
|
delivery_key=delivery_key,
|
|
)
|
|
if started:
|
|
logger.info(
|
|
"handoff_initiated_from_human account_id=%s message_id=%s",
|
|
account_id,
|
|
internet_message_id,
|
|
)
|
|
continue
|
|
external_recipients.append(recipient)
|
|
continue
|
|
await process_human_reply(
|
|
mongo_db,
|
|
account_id=account_id,
|
|
agent_email=recipient,
|
|
handoff=handoff,
|
|
parsed=parsed,
|
|
body_text=body_text,
|
|
body_html=body_html,
|
|
raw_mime=raw_mime,
|
|
delivery_key=delivery_key,
|
|
)
|
|
logger.info(
|
|
"handoff_reply_received account_id=%s handoff_id=%s message_id=%s",
|
|
account_id,
|
|
handoff["_id"],
|
|
internet_message_id,
|
|
)
|
|
|
|
if not external_recipients:
|
|
return f"250 2.0.0 Handoff message accepted as {delivery_key}"
|
|
recipients = external_recipients
|
|
|
|
message = {
|
|
**parsed,
|
|
"direction": "inbound",
|
|
"message_scope": MESSAGE_SCOPE_EXTERNAL,
|
|
"message_origin": MESSAGE_ORIGIN_EXTERNAL_CUSTOMER,
|
|
"generated_by": GENERATED_BY_CUSTOMER,
|
|
"envelope_sender": envelope.mail_from.lower(),
|
|
"envelope_recipients": recipients,
|
|
"delivery_key": delivery_key,
|
|
**id_fields,
|
|
"raw_mime": raw_mime,
|
|
"received_at": now,
|
|
"created_at": now,
|
|
"has_text_body": bool(body_text),
|
|
"has_html_body": bool(body_html),
|
|
}
|
|
await mongo_db.messages.update_one(
|
|
{"delivery_key": delivery_key},
|
|
{"$setOnInsert": message},
|
|
upsert=True,
|
|
)
|
|
persisted = await mongo_db.messages.find_one({"delivery_key": delivery_key})
|
|
if persisted is None:
|
|
raise RuntimeError("Inbound message was not persisted.")
|
|
|
|
if body_text or body_html:
|
|
reply_display = build_reply_display(body_text=body_text, body_html=body_html)
|
|
await mongo_db.messages.update_one(
|
|
{"_id": persisted["_id"]},
|
|
{
|
|
"$set": {
|
|
"body_text": body_text,
|
|
"body_html": body_html,
|
|
"new_reply_content": reply_display["new_reply_content"],
|
|
"quoted_content": reply_display["quoted_content"],
|
|
"has_quoted_text": reply_display["has_quoted_text"],
|
|
"snippet": reply_display["snippet"],
|
|
}
|
|
},
|
|
)
|
|
persisted["body_text"] = body_text
|
|
persisted["body_html"] = body_html
|
|
persisted["snippet"] = reply_display["snippet"]
|
|
persisted["new_reply_content"] = reply_display["new_reply_content"]
|
|
persisted["quoted_content"] = reply_display["quoted_content"]
|
|
persisted["has_quoted_text"] = reply_display["has_quoted_text"]
|
|
|
|
participants = participants_from_parsed(parsed)
|
|
for recipient in recipients:
|
|
account_id = envelope.account_ids[recipient]
|
|
resolution = await resolve_thread(
|
|
mongo_db,
|
|
account_id=account_id,
|
|
in_reply_to=parsed.get("in_reply_to"),
|
|
references=parsed.get("references", []),
|
|
subject=parsed.get("subject", ""),
|
|
participants=participants,
|
|
)
|
|
thread_id = await ensure_thread_for_message(
|
|
mongo_db,
|
|
account_id=account_id,
|
|
message=persisted,
|
|
resolution=resolution,
|
|
)
|
|
await mongo_db.email_threads.update_one(
|
|
{"_id": thread_id},
|
|
{"$set": {"thread_type": "EXTERNAL"}},
|
|
)
|
|
await mongo_db.messages.update_one(
|
|
{"_id": persisted["_id"]},
|
|
{
|
|
"$set": {
|
|
"message_scope": MESSAGE_SCOPE_EXTERNAL,
|
|
**id_fields,
|
|
"in_reply_to": canonical_message_id(parsed.get("in_reply_to")),
|
|
"references": parsed.get("references", []),
|
|
"raw_headers": parsed.get("raw_headers", {}),
|
|
"parent_message_id": resolution.parent_message_id,
|
|
"thread_relation": resolution.thread_relation,
|
|
"thread_resolution_method": resolution.thread_resolution_method,
|
|
}
|
|
},
|
|
)
|
|
mailbox_result = await mongo_db.mailbox_entries.update_one(
|
|
{
|
|
"account_id": account_id,
|
|
"message_id": persisted["_id"],
|
|
"mailbox": "inbox",
|
|
},
|
|
{
|
|
"$setOnInsert": {
|
|
"account_id": account_id,
|
|
"message_id": persisted["_id"],
|
|
"mailbox": "inbox",
|
|
"message_scope": MESSAGE_SCOPE_EXTERNAL,
|
|
"direction": "inbound",
|
|
"sender": parsed["sender"],
|
|
"recipients": parsed["to"],
|
|
"subject": parsed["subject"],
|
|
"snippet": persisted.get("snippet", parsed["snippet"]),
|
|
"attachments": parsed["attachments"],
|
|
"status": "received",
|
|
"is_read": False,
|
|
"is_starred": False,
|
|
"event_at": message_event_at(persisted),
|
|
"created_at": now,
|
|
"trashed_at": None,
|
|
"deletion_state": None,
|
|
},
|
|
"$set": {"thread_id": thread_id},
|
|
},
|
|
upsert=True,
|
|
)
|
|
log_original_external_message(
|
|
internal_id=persisted["_id"],
|
|
thread_id=thread_id,
|
|
internet_message_id=internet_message_id,
|
|
sender=sender_address,
|
|
recipient=recipient,
|
|
subject=parsed.get("subject"),
|
|
in_reply_to=canonical_message_id(parsed.get("in_reply_to")),
|
|
references=parsed.get("references", []),
|
|
raw_message_available=bool(persisted.get("raw_mime")),
|
|
)
|
|
log_smtp_message_received(
|
|
message_id=internet_message_id,
|
|
in_reply_to=parsed.get("in_reply_to"),
|
|
references=parsed.get("references", []),
|
|
sender=sender_address,
|
|
recipient=recipient,
|
|
subject=parsed.get("subject", ""),
|
|
resolved_scope=MESSAGE_SCOPE_EXTERNAL,
|
|
resolved_thread_id=thread_id,
|
|
resolution_method=resolution.thread_resolution_method,
|
|
stored_message_id=persisted["_id"],
|
|
mailbox_id="inbox",
|
|
)
|
|
logger.info(
|
|
"message_received message_id=%s internet_message_id=%s thread_id=%s "
|
|
"scope=EXTERNAL mailbox_entry_upserted=%s",
|
|
persisted["_id"],
|
|
internet_message_id,
|
|
thread_id,
|
|
mailbox_result.upserted_id is not None or mailbox_result.modified_count > 0,
|
|
)
|
|
try:
|
|
await maybe_trigger_handoff(
|
|
mongo_db,
|
|
account_id=account_id,
|
|
agent_email=recipient,
|
|
external_message=persisted,
|
|
external_thread_id=thread_id,
|
|
)
|
|
except Exception:
|
|
logger.exception(
|
|
"handoff_trigger_failed account_id=%s thread_id=%s message_id=%s",
|
|
account_id,
|
|
thread_id,
|
|
persisted["_id"],
|
|
)
|
|
except Exception:
|
|
logger.exception("Temporary failure accepting inbound message")
|
|
return "451 4.3.0 Temporary server failure; please retry"
|
|
return f"250 2.0.0 Message accepted as {delivery_key}"
|
|
|
|
|
|
class InboundSMTPServer:
|
|
def __init__(self):
|
|
self.server: asyncio.Server | None = None
|
|
|
|
async def start(self) -> None:
|
|
settings = get_settings()
|
|
loop = asyncio.get_running_loop()
|
|
self.server = await loop.create_server(
|
|
lambda: SMTP(
|
|
InboundHandler(),
|
|
data_size_limit=settings.inbound_max_bytes,
|
|
enable_SMTPUTF8=True,
|
|
decode_data=False,
|
|
),
|
|
host=settings.inbound_smtp_host,
|
|
port=settings.inbound_smtp_port,
|
|
)
|
|
|
|
async def stop(self) -> None:
|
|
if self.server is not None:
|
|
self.server.close()
|
|
await self.server.wait_closed()
|
|
self.server = None
|