diff --git a/apps/api/src/seskit_api/dependencies.py b/apps/api/src/seskit_api/dependencies.py index e798bc0..857142e 100644 --- a/apps/api/src/seskit_api/dependencies.py +++ b/apps/api/src/seskit_api/dependencies.py @@ -25,6 +25,7 @@ from seskit_core.security.sessions import SessionData, read_session from seskit_core.services import ( InboundProvisionerFactory, + InboundStoreFactory, ProviderFactory, ProvisionerFactory, get_default_project, @@ -33,7 +34,12 @@ touch_last_used, verify_api_key, ) -from seskit_provider_aws_ses import SESEventProvisioner, SESInboundProvisioner, SESProvider +from seskit_provider_aws_ses import ( + S3InboundStore, + SESEventProvisioner, + SESInboundProvisioner, + SESProvider, +) from sqlalchemy.ext.asyncio import AsyncSession @@ -113,6 +119,16 @@ def get_inbound_provisioner_factory() -> InboundProvisionerFactory: return SESInboundProvisioner +def get_inbound_store_factory() -> InboundStoreFactory: + """How a route reads a stored message back out of its bucket. + + Separate from the provisioner because reading mail and building the plumbing + for it are different capabilities with different permissions - a key that + can fetch an object does not need to be able to create a bucket. + """ + return S3InboundStore + + async def get_optional_user( request: Request, db: Annotated[AsyncSession, Depends(get_session)], diff --git a/apps/api/src/seskit_api/middleware/security.py b/apps/api/src/seskit_api/middleware/security.py index e4799d2..0ca630b 100644 --- a/apps/api/src/seskit_api/middleware/security.py +++ b/apps/api/src/seskit_api/middleware/security.py @@ -91,7 +91,13 @@ async def dispatch( response = await call_next(request) - response.headers["Content-Security-Policy"] = policy_for(nonce) + # A route that set its own policy keeps it. This is the dashboard's + # policy, written for pages that render here; applying it over a route + # that deliberately set a stricter one - a download of somebody else's + # file, which sets `sandbox` - would silently discard the stricter one + # and ship the looser, with nothing to say so. + if "Content-Security-Policy" not in response.headers: + response.headers["Content-Security-Policy"] = policy_for(nonce) response.headers["X-Content-Type-Options"] = "nosniff" response.headers["X-Frame-Options"] = "DENY" # same-origin rather than no-referrer: an unsubscribe token or an email diff --git a/apps/api/src/seskit_api/routes/v1/__init__.py b/apps/api/src/seskit_api/routes/v1/__init__.py index e86009d..5bc9f72 100644 --- a/apps/api/src/seskit_api/routes/v1/__init__.py +++ b/apps/api/src/seskit_api/routes/v1/__init__.py @@ -10,13 +10,14 @@ from fastapi import APIRouter -from seskit_api.routes.v1 import api_keys, domains, emails, events, webhooks +from seskit_api.routes.v1 import api_keys, domains, emails, events, inbound, webhooks router = APIRouter(prefix="/v1") router.include_router(api_keys.router) router.include_router(domains.router) router.include_router(emails.router) router.include_router(events.router) +router.include_router(inbound.router) router.include_router(webhooks.router) __all__ = ["router"] diff --git a/apps/api/src/seskit_api/routes/v1/inbound.py b/apps/api/src/seskit_api/routes/v1/inbound.py new file mode 100644 index 0000000..a406397 --- /dev/null +++ b/apps/api/src/seskit_api/routes/v1/inbound.py @@ -0,0 +1,333 @@ +"""``/v1/inbound`` - mail the project received. + +The read side of receiving. Nothing here writes: a message is recorded by the +worker as it arrives, and this only says what is there. + +Everything under ``/v1`` is scoped to the key's project in the query itself, so +an id belonging to another project resolves to a 404 and not a 403 - which +would confirm it exists. +""" + +from __future__ import annotations + +import re +from typing import Annotated, Any +from urllib.parse import quote + +from fastapi import APIRouter, Depends, Path, Query, Response +from seskit_core.config import Settings +from seskit_core.db import get_session +from seskit_core.email.parse import read_attachment +from seskit_core.errors import APIError, ErrorType +from seskit_core.models import InboundEmail +from seskit_core.models.base import utcnow +from seskit_core.services import InboundStoreFactory, get_connection, stored_credentials +from seskit_provider_aws_ses import EXPIRED_MESSAGE +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from seskit_api.dependencies import ( + APIContext, + get_app_settings, + get_inbound_store_factory, + require_api_key, +) +from seskit_api.routes.v1.api_keys import API_RESPONSES, apply_rate_limit_headers +from seskit_api.schemas.inbound import ( + InboundList, + InboundResponse, + detail_of, + summary_of, +) + +router = APIRouter(tags=["inbound"]) + +#: How many messages one page returns when the caller does not say. +DEFAULT_PAGE = 25 + +#: The most one page will ever return. A cap and not a suggestion, as on +#: ``/v1/emails``: without one a project with a year of mail can ask for all of +#: it in a single query. +MAX_PAGE = 100 + + +async def _owned(db: AsyncSession, inbound_id: str, project_id: str) -> InboundEmail | None: + """One received message, if it belongs to this project. + + Ownership is part of the query, so an id from another project is + indistinguishable from one that never existed. + """ + found: InboundEmail | None = await db.scalar( + select(InboundEmail).where( + InboundEmail.id == inbound_id, InboundEmail.project_id == project_id + ) + ) + return found + + +@router.get( + "/inbound/{inbound_id}", + response_model=InboundResponse, + responses=API_RESPONSES, + summary="Retrieve a received email", +) +async def get_inbound( + response: Response, + inbound_id: Annotated[ + str, Path(description="The id from the list, or from an `email.received` event.") + ], + db: Annotated[AsyncSession, Depends(get_session)], + context: Annotated[APIContext, Depends(require_api_key)], +) -> InboundResponse: + """One message, parsed. Bodies, headers, attachments and what SES concluded. + + The bodies are the sender's, not SESKit's. In particular the HTML is whatever + was sent, and has to be treated as hostile by anything that renders it. + """ + apply_rate_limit_headers(response, context) + + row = await _owned(db, inbound_id, context.project.id) + if row is None: + raise APIError(ErrorType.NOT_FOUND, "No received email with that id.") + return detail_of(row) + + +@router.get( + "/inbound", + response_model=InboundList, + responses=API_RESPONSES, + summary="List received emails", +) +async def list_inbound( + response: Response, + db: Annotated[AsyncSession, Depends(get_session)], + context: Annotated[APIContext, Depends(require_api_key)], + limit: Annotated[ + int, + Query(ge=1, le=MAX_PAGE, description="How many messages to return, newest first."), + ] = DEFAULT_PAGE, + starting_after: Annotated[ + str | None, + Query( + description=( + "Return messages older than this id - the last id from the previous " + "page. The id must belong to this project." + ), + ), + ] = None, + domain: Annotated[ + str | None, + Query(description="Only messages received for this domain."), + ] = None, +) -> InboundList: + """This key's project, newest first, without bodies. + + Paged by cursor for the reason ``/v1/emails`` is: ids sort in the order they + were created, so ``starting_after`` names a fixed point and stays correct + while mail arrives underneath it, where an offset would silently skip the + message that moved across the boundary. An unknown ``starting_after`` is a 404 + and not an empty page, because the comparison is lexical and an id from + another project would otherwise position a page of real messages by a value + the caller cannot see. + + Sorted by id and not by ``received_at``. A backlog means a message can be + recorded long after it arrived, so the two orders can differ; id order is the + one a cursor can be stable over, and ``received_at`` is on every row. + """ + apply_rate_limit_headers(response, context) + + query = select(InboundEmail).where(InboundEmail.project_id == context.project.id) + + if starting_after is not None: + cursor = await db.scalar( + select(InboundEmail.id).where( + InboundEmail.id == starting_after, + InboundEmail.project_id == context.project.id, + ) + ) + if cursor is None: + raise APIError(ErrorType.NOT_FOUND, "No received email with that id to page from.") + query = query.where(InboundEmail.id < cursor) + + if domain is not None: + query = query.where(InboundEmail.domain == domain.strip().lower()) + + # One more than asked for, so "is there another page?" is answered by the + # query that fetched this one. + rows = list(await db.scalars(query.order_by(InboundEmail.id.desc()).limit(limit + 1))) + + return InboundList( + data=[summary_of(row) for row in rows[:limit]], + has_more=len(rows) > limit, + ) + + +# ------------------------------------------------------------- downloads --- +# +# Everything below serves bytes a stranger wrote, from the API's own origin. That +# is the whole risk, and it is handled the same way for both routes: whatever the +# sender declared, the response is a download and not a document. A browser that +# is pointed at one must not render it, sniff it, run script in it, or keep it. + +#: Applied to every download. +#: +#: ``nosniff`` stops a browser deciding the bytes are HTML because they look like +#: it. The CSP ``sandbox`` is what is left if one renders it anyway: no script, no +#: forms, no same-origin access to anything. ``no-store`` because this is somebody's +#: private mail and should not sit in a shared cache. +DOWNLOAD_HEADERS = { + "X-Content-Type-Options": "nosniff", + "Content-Security-Policy": "sandbox; default-src 'none'", + "Cache-Control": "private, no-store", +} + +_BINARY_OK: dict[int | str, dict[str, Any]] = { + 200: { + "description": "The bytes.", + "content": {"application/octet-stream": {"schema": {"type": "string", "format": "binary"}}}, + } +} + +_NOT_SAFE = re.compile(r"[^A-Za-z0-9._ -]") + + +def content_disposition(filename: str) -> str: + """A ``Content-Disposition`` that is always a download, for any filename. + + Two forms, because old clients read the plain one and current ones read the + encoded one. The plain form is reduced to characters that cannot end the + quoted string or start a new header; the encoded form carries the name with + everything outside the unreserved set percent-encoded. + + Both are built from the same basename, so neither can carry a path or a name + that is only dots, and an empty name becomes ``attachment`` in both. A name + that came through the parser already has all of that - this does not rely on + it, because it is the last thing between a sender's string and a header, and + a client saving the file should not have to sanitise what it was given. + """ + base = filename.replace("\\", "/").rsplit("/", 1)[-1].strip(" .") or "attachment" + fallback = _NOT_SAFE.sub("_", base) + return f"attachment; filename=\"{fallback}\"; filename*=UTF-8''{quote(base, safe='')}" + + +async def _original( + db: AsyncSession, + row: InboundEmail, + factory: InboundStoreFactory, + settings: Settings, +) -> bytes: + """The stored message, or the right error for why not. + + The retention date is checked before anything is fetched. After it the original + is gone or about to be, and an S3 call to confirm costs a round trip that can + only produce the same answer - so a request past it is refused at once, and one + with no known expiry is left to the fetch to settle. + """ + if row.raw_expires_at is not None and row.raw_expires_at <= utcnow(): + raise APIError(ErrorType.NOT_FOUND, EXPIRED_MESSAGE) + + connection = await get_connection(db, row.project_id) + if connection is None or not connection.has_credentials: + raise APIError( + ErrorType.INVALID_REQUEST, + "This project is not connected to AWS, so the original message cannot be " + "fetched. Connect it on the AWS page.", + ) + + store = factory( + connection.region, stored_credentials(connection, secret_key=settings.SECRET_KEY) + ) + return await store.fetch_message(bucket=row.storage_bucket, key=row.storage_key) + + +@router.get( + "/inbound/{inbound_id}/attachments/{index}", + response_class=Response, + responses={**API_RESPONSES, **_BINARY_OK}, + summary="Download an attachment", +) +async def download_attachment( + inbound_id: Annotated[str, Path(description="The id of the received message.")], + index: Annotated[ + int, + Path(ge=0, description="The attachment's `index` from the message's `attachments`."), + ], + db: Annotated[AsyncSession, Depends(get_session)], + context: Annotated[APIContext, Depends(require_api_key)], + settings: Annotated[Settings, Depends(get_app_settings)], + factory: Annotated[InboundStoreFactory, Depends(get_inbound_store_factory)], +) -> Response: + """One attachment's bytes, fetched from the stored message. + + **Always served as `application/octet-stream`**, whatever the sender declared, + with `Content-Disposition: attachment`, `nosniff` and a sandboxing CSP. The + declared type is on the message's `attachments` for you to read; it is never + one this endpoint believes. These are files a stranger sent, served from your + API's origin. + + Needs the stored message, so it stops working when retention removes it. The + parsed message does not. + """ + row = await _owned(db, inbound_id, context.project.id) + if row is None: + raise APIError(ErrorType.NOT_FOUND, "No received email with that id.") + if index >= len(row.attachments): + # Before the fetch: a request for an attachment that does not exist should + # not cost a download of the message to find that out. + raise APIError(ErrorType.NOT_FOUND, "That message has no attachment with that index.") + + raw = await _original(db, row, factory, settings) + content = read_attachment(raw, index) + if content is None: + raise APIError(ErrorType.NOT_FOUND, "That message has no attachment with that index.") + + out = Response( + content=content.data, + media_type="application/octet-stream", + headers={**DOWNLOAD_HEADERS, "Content-Disposition": content_disposition(content.filename)}, + ) + apply_rate_limit_headers(out, context) + return out + + +@router.get( + "/inbound/{inbound_id}/raw", + response_class=Response, + responses={ + **API_RESPONSES, + 200: { + "description": "The original message, byte for byte.", + "content": {"message/rfc822": {"schema": {"type": "string", "format": "binary"}}}, + }, + }, + summary="Download the original message", +) +async def download_raw( + inbound_id: Annotated[str, Path(description="The id of the received message.")], + db: Annotated[AsyncSession, Depends(get_session)], + context: Annotated[APIContext, Depends(require_api_key)], + settings: Annotated[Settings, Depends(get_app_settings)], + factory: Annotated[InboundStoreFactory, Depends(get_inbound_store_factory)], +) -> Response: + """The message exactly as SES received it, as an `.eml` file. + + Unparsed and unmodified: every header, every part, every attachment. Useful for + debugging a message that did not parse as expected, and for handing to another + tool. It is the sender's bytes, served as a download with the same protections + as an attachment. + + Needs the stored message, so it stops working when retention removes it. + """ + row = await _owned(db, inbound_id, context.project.id) + if row is None: + raise APIError(ErrorType.NOT_FOUND, "No received email with that id.") + + raw = await _original(db, row, factory, settings) + + out = Response( + content=raw, + media_type="message/rfc822", + headers={**DOWNLOAD_HEADERS, "Content-Disposition": content_disposition(f"{row.id}.eml")}, + ) + apply_rate_limit_headers(out, context) + return out diff --git a/apps/api/src/seskit_api/schemas/inbound.py b/apps/api/src/seskit_api/schemas/inbound.py new file mode 100644 index 0000000..242bcf8 --- /dev/null +++ b/apps/api/src/seskit_api/schemas/inbound.py @@ -0,0 +1,260 @@ +"""Response models for received mail (inbound email). + +Two shapes. The list returns a *summary* - who, what, when, how big - and no +bodies, because a page of a hundred messages with their HTML in it is a response +nobody asked for. The single message returns everything that was parsed. + +Both are built by hand from the stored row rather than with ``from_attributes``. +The row stores things in shapes that suit the database - headers as pairs, the +envelope recipients under a name that only makes sense beside ``to_addresses`` - +and the public shape is a contract that should not move when the table does. + +``from`` is a Python keyword, so the field is ``from_address`` with a +serialization alias, exactly as on ``EmailResponse``. + +Everything in here was written by somebody else. Nothing is trusted, and the +descriptions say so where a consumer might otherwise assume it. +""" + +from __future__ import annotations + +from datetime import datetime + +from pydantic import BaseModel, Field +from seskit_core.models import InboundEmail +from seskit_core.models.base import utcnow + + +class InboundVerdicts(BaseModel): + """What SES concluded about the message. Recorded, never acted on.""" + + spf: str | None = Field( + default=None, + description=( + "`PASS`, `FAIL`, `GRAY` or `PROCESSING_FAILED`. Null means SES did not " + "say, which is not the same as `PASS`." + ), + examples=["PASS"], + ) + dkim: str | None = Field( + default=None, + description=( + "The DKIM verdict, in the same values as `spf`. `GRAY` means the message " + "was not signed or the signing domain does not match the sender's." + ), + examples=["GRAY"], + ) + dmarc: str | None = Field( + default=None, description="The DMARC verdict, in the same values as `spf`." + ) + dmarc_policy: str | None = Field( + default=None, + description=( + "The sender domain's published policy - `none`, `quarantine` or `reject` - " + "and present only when DMARC failed." + ), + ) + spam: str | None = Field(default=None, description="The spam verdict, in the same values.") + virus: str | None = Field(default=None, description="The virus verdict, in the same values.") + + +class InboundAttachment(BaseModel): + """One attachment, described. The bytes are fetched separately.""" + + index: int = Field( + description=( + "Its position among the message's attachments. Pass it to " + "`GET /v1/inbound/{id}/attachments/{index}`." + ), + examples=[0], + ) + filename: str = Field( + description=( + "Reduced to a plain name: no path, no control characters. The sender " + "chose it, so treat it as untrusted text." + ), + examples=["report.pdf"], + ) + content_type: str = Field( + description=( + "What the sender *declared*. It is not checked, and a download is always " + "served as `application/octet-stream` regardless." + ), + examples=["application/pdf"], + ) + size: int = Field(description="Size in bytes, after decoding.", examples=[48213]) + inline: bool = Field( + description="Whether it is part of the message's layout, such as an embedded image." + ) + content_id: str | None = Field( + default=None, + description="The `Content-ID` an HTML body refers to as `cid:`, if it has one.", + ) + + +class InboundHeader(BaseModel): + """One header as the sender wrote it.""" + + name: str = Field(description="The header name.", examples=["Received-SPF"]) + value: str = Field( + description="The raw value, unfolded and not decoded. Untrusted.", + examples=["pass"], + ) + + +class InboundSummary(BaseModel): + """A received message, without its body. What the list returns.""" + + id: str = Field( + description="Opaque and stable, prefixed `inbound_`.", + examples=["inbound_01J8XQ2K3M4N5P6Q7R8S9T0V1W"], + ) + domain: str = Field( + description="The domain the message was received for.", examples=["example.com"] + ) + from_address: str = Field( + serialization_alias="from", + description=( + "The address in the `From` header, lower-cased. Empty if there was none. " + "Anybody can write anything here: see `verdicts` for what SES checked." + ), + examples=["ada@other.org"], + ) + from_name: str = Field(description="The display name in `From`, if any.", examples=["Ada"]) + to: list[str] = Field(description="The addresses in the `To` header.") + cc: list[str] = Field(description="The addresses in the `Cc` header. Empty if none.") + subject: str = Field(description="The subject, decoded and on one line.") + received_at: datetime = Field(description="When SES received the message. UTC.") + size: int = Field(description="Size of the whole message in bytes.", examples=[4821]) + attachment_count: int = Field(description="How many attachments it has.", examples=[0]) + verdicts: InboundVerdicts = Field(description="What SES concluded about the message.") + truncated: bool = Field( + description=( + "Whether a limit was hit while reading it, so something is missing from " + "this representation. The original has all of it." + ) + ) + parse_failed: bool = Field( + description=( + "Whether the message could not be read at all. Everything else is empty; " + "the original is available through `/raw` until it expires." + ) + ) + raw_available: bool = Field( + description=( + "Whether the original is expected to still be in storage. Retention " + "removes it at a day boundary, so a message just past `raw_expires_at` " + "may already be gone and one just before it should not be." + ) + ) + raw_expires_at: datetime | None = Field( + default=None, + description=( + "When retention will have removed the original and its attachments. " + "Null when unknown. The parsed message below stays regardless. UTC." + ), + ) + + +class InboundResponse(InboundSummary): + """One received message, in full.""" + + text_body: str = Field( + serialization_alias="text", + description="The plain-text body. Empty when the message had none.", + ) + html_body: str = Field( + serialization_alias="html", + description=( + "The HTML body, exactly as sent. **Untrusted.** It may contain script, " + "tracking images and links that lie; do not render it in a page that " + "carries your own credentials." + ), + ) + envelope_from: str = Field( + description=("The SMTP envelope sender. Empty for a bounce. It need not match `from`."), + ) + recipients: list[str] = Field( + description=( + "Who the message was addressed to on the wire - the addresses that " + "matched your rule. A blind copy is here and in neither `to` nor `cc`." + ) + ) + reply_to: list[str] = Field(description="The addresses in the `Reply-To` header.") + message_id: str = Field( + description="The `Message-ID` header, without angle brackets. Empty if none.", + ) + in_reply_to: str = Field( + description="The message this one answers, without angle brackets. Empty if none.", + ) + references: list[str] = Field( + description="The ids of the thread it belongs to, oldest first, without brackets." + ) + attachments: list[InboundAttachment] = Field( + description="Described in the order `/attachments/{index}` counts them." + ) + headers: list[InboundHeader] = Field( + description="Every header as sent, in order, up to a limit. See `truncated`." + ) + + +class InboundList(BaseModel): + """One page of a project's received messages, newest first.""" + + data: list[InboundSummary] = Field( + description="The page, ordered by id descending - newest first." + ) + has_more: bool = Field( + description=( + "Whether another page exists. Pass the last id in `data` as " + "`starting_after` to fetch it." + ), + examples=[False], + ) + + +def _raw_available(row: InboundEmail) -> bool: + return row.raw_expires_at is None or row.raw_expires_at > utcnow() + + +def summary_of(row: InboundEmail) -> InboundSummary: + return InboundSummary( + id=row.id, + domain=row.domain, + from_address=row.from_address, + from_name=row.from_name, + to=row.to_addresses, + cc=row.cc_addresses, + subject=row.subject, + received_at=row.received_at, + size=row.size_bytes, + attachment_count=len(row.attachments), + verdicts=InboundVerdicts( + spf=row.spf_verdict, + dkim=row.dkim_verdict, + dmarc=row.dmarc_verdict, + dmarc_policy=row.dmarc_policy, + spam=row.spam_verdict, + virus=row.virus_verdict, + ), + truncated=row.truncated, + parse_failed=row.parse_failed, + raw_available=_raw_available(row), + raw_expires_at=row.raw_expires_at, + ) + + +def detail_of(row: InboundEmail) -> InboundResponse: + return InboundResponse( + **summary_of(row).model_dump(), + text_body=row.text_body, + html_body=row.html_body, + envelope_from=row.envelope_from, + recipients=row.envelope_to, + reply_to=row.reply_to, + message_id=row.message_id_header, + in_reply_to=row.in_reply_to, + references=row.reference_ids, + attachments=[InboundAttachment(**attachment) for attachment in row.attachments], + headers=[InboundHeader(name=name, value=value) for name, value in row.headers], + ) diff --git a/apps/worker/src/seskit_worker/inbound.py b/apps/worker/src/seskit_worker/inbound.py index db244b7..05ef941 100644 --- a/apps/worker/src/seskit_worker/inbound.py +++ b/apps/worker/src/seskit_worker/inbound.py @@ -44,6 +44,7 @@ MalformedEnvelope, NotAReceipt, Outcome, + event_for, is_recorded, locate, parse_received, @@ -57,10 +58,11 @@ NotificationQueue, QueuedNotification, ) +from seskit_core.services import pending_delivery_ids from seskit_core.services.inbound import distinct_inbound_queues from seskit_provider_aws_ses import S3InboundStore -from seskit_worker.events import SessionFactory, build_queue +from seskit_worker.events import Enqueue, SessionFactory, _enqueue_via, build_queue logger = get_logger(__name__) @@ -83,6 +85,7 @@ async def poll_inbound( build: Callable[[str, str, AWSCredentials], NotificationQueue] | None = None, store_builder: StoreBuilder | None = None, session_factory: SessionFactory | None = None, + enqueue: Enqueue | None = None, ) -> int: """Drain every inbound queue this instance has. Returns messages recorded. @@ -97,6 +100,7 @@ async def poll_inbound( build = build or build_queue store_builder = store_builder or build_store factory = session_factory or get_session_factory() + enqueue = enqueue or _enqueue_via(ctx) recorded = 0 async with factory() as session: @@ -111,6 +115,7 @@ async def poll_inbound( session_factory=factory, project_ids=inbox.project_ids, retention_days=settings.INBOUND_RETENTION_DAYS, + enqueue=enqueue, max_batches=settings.EVENT_POLL_MAX_BATCHES, wait_seconds=settings.EVENT_POLL_WAIT_SECONDS, visibility_timeout=settings.EVENT_VISIBILITY_TIMEOUT_SECONDS, @@ -135,6 +140,7 @@ async def drain_inbound( max_batches: int, wait_seconds: int, visibility_timeout: int, + enqueue: Enqueue | None = None, ) -> int: """Read batches until the queue is empty or the budget runs out. @@ -161,6 +167,7 @@ async def drain_inbound( session_factory=session_factory, project_ids=project_ids, retention_days=retention_days, + enqueue=enqueue, ) except Exception as exc: # Left on the queue, and the rest of the batch carries on. The @@ -187,6 +194,7 @@ async def handle_inbound( session_factory: SessionFactory, project_ids: Collection[str], retention_days: int, + enqueue: Enqueue | None = None, ) -> bool: """Process one announcement. Returns whether a message was recorded. @@ -249,7 +257,7 @@ async def handle_inbound( raise async with session_factory() as session: - outcome, _ = await record_received( + outcome, row = await record_received( session, identity, received, @@ -257,8 +265,23 @@ async def handle_inbound( provider_event_id=event_id, retention_days=retention_days, ) + # Read before the commit closes the session, so the ids survive. + delivery_ids: list[str] = [] + if enqueue is not None and row is not None and outcome is Outcome.RECORDED: + event = await event_for(session, row.id) + if event is not None: + delivery_ids = await pending_delivery_ids(session, event.id) await session.commit() + for delivery_id in delivery_ids: + # Latency only: the delivery row is what makes the webhook durable, and + # the sweep attempts anything the enqueue lost within the minute. A + # failure here must not keep the message on the queue - it is recorded. + try: + await enqueue(delivery_id) # type: ignore[misc] + except Exception as exc: + logger.warning("inbound_enqueue_failed", error=type(exc).__name__) + if outcome.is_settled: await queue.delete(notification) diff --git a/docs/design/security-model.md b/docs/design/security-model.md index 00cff00..3411831 100644 --- a/docs/design/security-model.md +++ b/docs/design/security-model.md @@ -149,6 +149,52 @@ later without a flag day. after a logout cannot re-display the previous user's dashboard from the browser cache. +## Received mail is hostile input + +Everything in a message somebody sent to you was written by somebody who may mean +you harm. It is a third untrusted boundary, and unlike the two above it is one +SESKit serves *from*. + +**Announcements are believed only if they are ours.** Nothing in an SES +notification is secret: the message id is written into the headers of every +message SES delivers. So the project a message belongs to comes from the +recipient's domain, and is accepted only if it is one of the projects the queue +was read for and the bucket is the one SESKit created. Every way of being wrong +is dropped the same way, before anything is downloaded, so a probe learns nothing +from the difference. + +**A download is a download, whatever the sender declared.** Attachments and the +original message are always served with `Content-Disposition: attachment`, +`X-Content-Type-Options: nosniff` and `Content-Security-Policy: sandbox; +default-src 'none'`, and attachments as `application/octet-stream` regardless of +the type they claim. The filename is the sender's string and is reduced to a +plain basename in both forms of the header, so it cannot carry a path or end the +header. A test sends a file that declares itself HTML and carries script, and a +name containing a CRLF-delimited `Set-Cookie`, and checks the bytes of the +response. + +**The API does not clean what it returns.** `html` and the headers come back +exactly as sent, and the field descriptions say to treat them as hostile. A +sanitiser here would be a second place for a consumer to be wrong. + +**Verdicts are recorded and never acted on.** Amazon SES takes no action on a +failed SPF, DKIM or DMARC check, and neither does SESKit; whether to trust a +message is the reader's decision. A verdict that was not given is stored and +returned as `null`, never as a pass. + +**Storage.** The bucket is private, encrypted, accessible over TLS only, and +writable by receipt rules from this account named `seskit-*` and nothing else. The +original is removed after the retention period; the parsed message is not, so a +message past retention is still readable but its attachments can no longer be +downloaded. + +**Credentials are the project's own.** A message is read with the key of the +project that received it, in its own region. There is no instance-wide key. + +**Nothing from the message is logged.** Sender, recipients, subject and bodies +stay out of logs; an error raised while handling a message is recorded by its type +alone, because its text can carry a fragment of the mail. + ## Response headers Set by one middleware on every response, because the failure mode of @@ -156,7 +202,7 @@ per-route security is a route somebody forgot. | Header | | |---|---| -| `Content-Security-Policy` | `default-src 'self'`, with a per-request nonce for the two inline theme scripts. No `unsafe-inline`, no `unsafe-eval` | +| `Content-Security-Policy` | `default-src 'self'`, with a per-request nonce for the two inline theme scripts. No `unsafe-inline`, no `unsafe-eval`. A route that sets a stricter policy of its own keeps it - downloads of received mail do - and the middleware does not overwrite it | | `X-Frame-Options` / `frame-ancestors` | `DENY` / `'none'`. Every destructive action on the dashboard is a one-click form, which is what clickjacking needs | | `form-action 'self'` | The directive framing protection has no answer for: injected markup posting a CSRF token to another origin | | `X-Content-Type-Options` | `nosniff` | @@ -209,6 +255,9 @@ Stated plainly rather than left to be discovered: - **No HTTP API for the suppression list.** Addresses are suppressed automatically and managed on the dashboard; there is no `/v1` endpoint to read or bulk-load the list. See [suppression](../guides/suppression.md). +- **Received mail is not quarantined.** Amazon SES scans it for spam and viruses + and SESKit records the verdicts, but neither blocks a message from being + stored or served. SESKit does not scan attachments itself. - **No RBAC.** An account owns its projects; there are no roles or team members. - **No per-IP rate limit.** Limits are per project for the API and per account diff --git a/docs/guides/webhooks.md b/docs/guides/webhooks.md index 85521e5..f001a27 100644 --- a/docs/guides/webhooks.md +++ b/docs/guides/webhooks.md @@ -29,6 +29,11 @@ The body is the same normalised event the dashboard stores: } ``` +!!! note "Received mail arrives here too" + Mail sent *to* a domain you receive for produces an `email.received` event. + It has no `email_id` — it carries `inbound_id` — and describes the message + without carrying it. See the [events reference](../reference/events.md#emailreceived). + !!! important "Webhooks need delivery events set up first" Without a configuration set, Amazon SES publishes nothing, so there is nothing to forward. See [delivery events](delivery-events.md). diff --git a/docs/reference/api.md b/docs/reference/api.md index 30128ff..3175a08 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -59,6 +59,10 @@ Per project, not per key. Every response carries `X-RateLimit-Limit`, | `POST` | [`/v1/emails`](#post-v1emails) | Send an email | | `GET` | [`/v1/emails`](#get-v1emails) | List emails | | `GET` | [`/v1/emails/{email_id}`](#get-v1emailsemail_id) | Retrieve an email | +| `GET` | [`/v1/inbound`](#get-v1inbound) | List received emails | +| `GET` | [`/v1/inbound/{inbound_id}`](#get-v1inboundinbound_id) | Retrieve a received email | +| `GET` | [`/v1/inbound/{inbound_id}/attachments/{index}`](#get-v1inboundinbound_idattachmentsindex) | Download an attachment | +| `GET` | [`/v1/inbound/{inbound_id}/raw`](#get-v1inboundinbound_idraw) | Download the original message | | `GET` | [`/v1/webhooks`](#get-v1webhooks) | List webhook endpoints | | `GET` | [`/v1/webhooks/{endpoint_id}/deliveries`](#get-v1webhooksendpoint_iddeliveries) | List recent webhook deliveries | @@ -203,6 +207,136 @@ rather than a 403 - which would confirm the id exists. | `429` | Rate limit exceeded. | +--- + +## `GET /v1/inbound` + +**List received emails** + +This key's project, newest first, without bodies. + +Paged by cursor for the reason ``/v1/emails`` is: ids sort in the order they +were created, so ``starting_after`` names a fixed point and stays correct +while mail arrives underneath it, where an offset would silently skip the +message that moved across the boundary. An unknown ``starting_after`` is a 404 +and not an empty page, because the comparison is lexical and an id from +another project would otherwise position a page of real messages by a value +the caller cannot see. + +Sorted by id and not by ``received_at``. A backlog means a message can be +recorded long after it arrived, so the two orders can differ; id order is the +one a cursor can be stable over, and ``received_at`` is on every row. + +**Parameters** + +| Name | In | | | +|---|---|---|---| +| `limit` | query | optional | How many messages to return, newest first. | +| `starting_after` | query | optional | Return messages older than this id - the last id from the previous page. The id must belong to this project. | +| `domain` | query | optional | Only messages received for this domain. | + +**Responses** + +| Status | | +|---|---| +| `200` | Successful Response | +| `401` | Invalid or missing API key. | +| `422` | Validation Error | +| `429` | Rate limit exceeded. | + + +--- + +## `GET /v1/inbound/{inbound_id}` + +**Retrieve a received email** + +One message, parsed. Bodies, headers, attachments and what SES concluded. + +The bodies are the sender's, not SESKit's. In particular the HTML is whatever +was sent, and has to be treated as hostile by anything that renders it. + +**Parameters** + +| Name | In | | | +|---|---|---|---| +| `inbound_id` | path | required | The id from the list, or from an `email.received` event. | + +**Responses** + +| Status | | +|---|---| +| `200` | Successful Response | +| `401` | Invalid or missing API key. | +| `422` | Validation Error | +| `429` | Rate limit exceeded. | + + +--- + +## `GET /v1/inbound/{inbound_id}/attachments/{index}` + +**Download an attachment** + +One attachment's bytes, fetched from the stored message. + +**Always served as `application/octet-stream`**, whatever the sender declared, +with `Content-Disposition: attachment`, `nosniff` and a sandboxing CSP. The +declared type is on the message's `attachments` for you to read; it is never +one this endpoint believes. These are files a stranger sent, served from your +API's origin. + +Needs the stored message, so it stops working when retention removes it. The +parsed message does not. + +**Parameters** + +| Name | In | | | +|---|---|---|---| +| `inbound_id` | path | required | The id of the received message. | +| `index` | path | required | The attachment's `index` from the message's `attachments`. | + +**Responses** + +| Status | | +|---|---| +| `200` | The bytes. | +| `401` | Invalid or missing API key. | +| `422` | Validation Error | +| `429` | Rate limit exceeded. | + + +--- + +## `GET /v1/inbound/{inbound_id}/raw` + +**Download the original message** + +The message exactly as SES received it, as an `.eml` file. + +Unparsed and unmodified: every header, every part, every attachment. Useful for +debugging a message that did not parse as expected, and for handing to another +tool. It is the sender's bytes, served as a download with the same protections +as an attachment. + +Needs the stored message, so it stops working when retention removes it. + +**Parameters** + +| Name | In | | | +|---|---|---|---| +| `inbound_id` | path | required | The id of the received message. | + +**Responses** + +| Status | | +|---|---| +| `200` | The original message, byte for byte. | +| `401` | Invalid or missing API key. | +| `422` | Validation Error | +| `429` | Rate limit exceeded. | + + --- ## `GET /v1/webhooks` diff --git a/docs/reference/events.md b/docs/reference/events.md index babbdfc..ade46cf 100644 --- a/docs/reference/events.md +++ b/docs/reference/events.md @@ -7,7 +7,7 @@ it forwards to your [webhooks](../guides/webhooks.md). ```text email.sent email.delivered email.bounced email.suppressed -email.opened email.clicked email.complained +email.opened email.clicked email.complained email.received ``` | Type | Emitted when | @@ -19,6 +19,7 @@ email.opened email.clicked email.complained | `email.opened` | The tracking pixel loaded — only with [tracking on](../guides/delivery-events.md#open-and-click-tracking) | | `email.clicked` | A rewritten link was followed — tracking only | | `email.suppressed` | SESKit added an address to the [suppression list](../guides/suppression.md) | +| `email.received` | Mail arrived for one of your domains — see [below](#emailreceived) | SES emits more types than these — `Reject`, `Rendering Failure`, `DeliveryDelay` — which SESKit records but does not currently forward. @@ -42,8 +43,8 @@ SES emits more types than these — `Reject`, `Rendering Failure`, | Field | | |---|---| | `id` | Unique per event. **Deduplicate on this** — delivery is at-least-once | -| `type` | One of the seven above | -| `email_id` | The message this is about; matches `GET /v1/emails/{id}` | +| `type` | One of the eight above | +| `email_id` | The message this is about; matches `GET /v1/emails/{id}`. **Absent on `email.received`**, which carries `inbound_id` instead | | `created_at` | When the event *happened*, not when SESKit heard about it | | `data` | Type-dependent. Always includes `to` | @@ -96,6 +97,70 @@ anything an hour ago. An address already on the list produces no second event, so a repeat bounce is not something your application has to deduplicate. +## `email.received` + +Raised when mail arrives for a domain you have receiving turned on for, and +readable through the [`/v1/inbound` endpoints](api.md#get-v1inbound). It is about a message somebody sent **to you**, so it has no +`email_id`: it carries `inbound_id`, which matches `GET /v1/inbound/{id}`. + +```json +{ + "id": "evt_01J8XQ...", + "type": "email.received", + "inbound_id": "inbound_01J8XQ...", + "created_at": "2026-10-05T10:00:00+00:00", + "data": { + "domain": "example.com", + "from": "ada@other.org", + "from_name": "Ada Lovelace", + "to": ["support@example.com"], + "cc": [], + "recipients": ["support@example.com"], + "subject": "Hello there", + "received_at": "2026-10-05T10:00:00+00:00", + "size": 4821, + "attachment_count": 1, + "verdicts": { + "spf": "PASS", + "dkim": "GRAY", + "dmarc": "PASS", + "spam": "PASS", + "virus": null + }, + "truncated": false, + "parse_failed": false + } +} +``` + +**It describes the message and does not carry it.** There is no body, no +headers and no attachment in the payload, however large the mail is. Fetch the +message with `GET /v1/inbound/{inbound_id}` — that is the call that returns the +bodies, the headers and the attachments. + +| Field | | +|---|---| +| `inbound_id` | The received message. Use it to fetch the rest | +| `from` | The address in the `From` header. Anybody can write anything there — see `verdicts` | +| `to` | The `To` header | +| `recipients` | Who the message was addressed to **on the wire** — the address that actually received it. A blind copy is in this list and in neither `to` nor `cc` | +| `verdicts` | What Amazon SES concluded: `PASS`, `FAIL`, `GRAY` or `PROCESSING_FAILED`. SESKit records them and acts on none | +| `truncated` | A limit was hit while reading the message, so something is missing from the stored copy | +| `parse_failed` | Nothing could be read. The message still exists and `GET /v1/inbound/{id}/raw` has the original | + +**A verdict that is `null` means Amazon SES did not say.** It is not a pass. +Filter on spam or failed authentication only after checking for it, or an +unchecked message will read as a clean one. + +`created_at` is when the message was *received*, not when SESKit processed it. + +Delivery is at-least-once, as for every other event, so **deduplicate on `id`**. +A message that is announced twice produces one event. + +Every active endpoint receives it, as with every other type. If yours checks +`type` against a fixed list, add `email.received` before you turn receiving on. +Nothing is sent until you do. + ## Timestamps `created_at` is when the event occurred at the provider, not when SESKit diff --git a/docs/reference/openapi.json b/docs/reference/openapi.json index e4fbe89..52d912b 100644 --- a/docs/reference/openapi.json +++ b/docs/reference/openapi.json @@ -565,6 +565,543 @@ "title": "HealthResponse", "type": "object" }, + "InboundAttachment": { + "description": "One attachment, described. The bytes are fetched separately.", + "properties": { + "content_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "description": "The `Content-ID` an HTML body refers to as `cid:`, if it has one.", + "title": "Content Id" + }, + "content_type": { + "description": "What the sender *declared*. It is not checked, and a download is always served as `application/octet-stream` regardless.", + "examples": [ + "application/pdf" + ], + "title": "Content Type", + "type": "string" + }, + "filename": { + "description": "Reduced to a plain name: no path, no control characters. The sender chose it, so treat it as untrusted text.", + "examples": [ + "report.pdf" + ], + "title": "Filename", + "type": "string" + }, + "index": { + "description": "Its position among the message's attachments. Pass it to `GET /v1/inbound/{id}/attachments/{index}`.", + "examples": [ + 0 + ], + "title": "Index", + "type": "integer" + }, + "inline": { + "description": "Whether it is part of the message's layout, such as an embedded image.", + "title": "Inline", + "type": "boolean" + }, + "size": { + "description": "Size in bytes, after decoding.", + "examples": [ + 48213 + ], + "title": "Size", + "type": "integer" + } + }, + "required": [ + "index", + "filename", + "content_type", + "size", + "inline" + ], + "title": "InboundAttachment", + "type": "object" + }, + "InboundHeader": { + "description": "One header as the sender wrote it.", + "properties": { + "name": { + "description": "The header name.", + "examples": [ + "Received-SPF" + ], + "title": "Name", + "type": "string" + }, + "value": { + "description": "The raw value, unfolded and not decoded. Untrusted.", + "examples": [ + "pass" + ], + "title": "Value", + "type": "string" + } + }, + "required": [ + "name", + "value" + ], + "title": "InboundHeader", + "type": "object" + }, + "InboundList": { + "description": "One page of a project's received messages, newest first.", + "properties": { + "data": { + "description": "The page, ordered by id descending - newest first.", + "items": { + "$ref": "#/components/schemas/InboundSummary" + }, + "title": "Data", + "type": "array" + }, + "has_more": { + "description": "Whether another page exists. Pass the last id in `data` as `starting_after` to fetch it.", + "examples": [ + false + ], + "title": "Has More", + "type": "boolean" + } + }, + "required": [ + "data", + "has_more" + ], + "title": "InboundList", + "type": "object" + }, + "InboundResponse": { + "description": "One received message, in full.", + "properties": { + "attachment_count": { + "description": "How many attachments it has.", + "examples": [ + 0 + ], + "title": "Attachment Count", + "type": "integer" + }, + "attachments": { + "description": "Described in the order `/attachments/{index}` counts them.", + "items": { + "$ref": "#/components/schemas/InboundAttachment" + }, + "title": "Attachments", + "type": "array" + }, + "cc": { + "description": "The addresses in the `Cc` header. Empty if none.", + "items": { + "type": "string" + }, + "title": "Cc", + "type": "array" + }, + "domain": { + "description": "The domain the message was received for.", + "examples": [ + "example.com" + ], + "title": "Domain", + "type": "string" + }, + "envelope_from": { + "description": "The SMTP envelope sender. Empty for a bounce. It need not match `from`.", + "title": "Envelope From", + "type": "string" + }, + "from": { + "description": "The address in the `From` header, lower-cased. Empty if there was none. Anybody can write anything here: see `verdicts` for what SES checked.", + "examples": [ + "ada@other.org" + ], + "title": "From", + "type": "string" + }, + "from_name": { + "description": "The display name in `From`, if any.", + "examples": [ + "Ada" + ], + "title": "From Name", + "type": "string" + }, + "headers": { + "description": "Every header as sent, in order, up to a limit. See `truncated`.", + "items": { + "$ref": "#/components/schemas/InboundHeader" + }, + "title": "Headers", + "type": "array" + }, + "html": { + "description": "The HTML body, exactly as sent. **Untrusted.** It may contain script, tracking images and links that lie; do not render it in a page that carries your own credentials.", + "title": "Html", + "type": "string" + }, + "id": { + "description": "Opaque and stable, prefixed `inbound_`.", + "examples": [ + "inbound_01J8XQ2K3M4N5P6Q7R8S9T0V1W" + ], + "title": "Id", + "type": "string" + }, + "in_reply_to": { + "description": "The message this one answers, without angle brackets. Empty if none.", + "title": "In Reply To", + "type": "string" + }, + "message_id": { + "description": "The `Message-ID` header, without angle brackets. Empty if none.", + "title": "Message Id", + "type": "string" + }, + "parse_failed": { + "description": "Whether the message could not be read at all. Everything else is empty; the original is available through `/raw` until it expires.", + "title": "Parse Failed", + "type": "boolean" + }, + "raw_available": { + "description": "Whether the original is expected to still be in storage. Retention removes it at a day boundary, so a message just past `raw_expires_at` may already be gone and one just before it should not be.", + "title": "Raw Available", + "type": "boolean" + }, + "raw_expires_at": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "description": "When retention will have removed the original and its attachments. Null when unknown. The parsed message below stays regardless. UTC.", + "title": "Raw Expires At" + }, + "received_at": { + "description": "When SES received the message. UTC.", + "format": "date-time", + "title": "Received At", + "type": "string" + }, + "recipients": { + "description": "Who the message was addressed to on the wire - the addresses that matched your rule. A blind copy is here and in neither `to` nor `cc`.", + "items": { + "type": "string" + }, + "title": "Recipients", + "type": "array" + }, + "references": { + "description": "The ids of the thread it belongs to, oldest first, without brackets.", + "items": { + "type": "string" + }, + "title": "References", + "type": "array" + }, + "reply_to": { + "description": "The addresses in the `Reply-To` header.", + "items": { + "type": "string" + }, + "title": "Reply To", + "type": "array" + }, + "size": { + "description": "Size of the whole message in bytes.", + "examples": [ + 4821 + ], + "title": "Size", + "type": "integer" + }, + "subject": { + "description": "The subject, decoded and on one line.", + "title": "Subject", + "type": "string" + }, + "text": { + "description": "The plain-text body. Empty when the message had none.", + "title": "Text", + "type": "string" + }, + "to": { + "description": "The addresses in the `To` header.", + "items": { + "type": "string" + }, + "title": "To", + "type": "array" + }, + "truncated": { + "description": "Whether a limit was hit while reading it, so something is missing from this representation. The original has all of it.", + "title": "Truncated", + "type": "boolean" + }, + "verdicts": { + "$ref": "#/components/schemas/InboundVerdicts", + "description": "What SES concluded about the message." + } + }, + "required": [ + "id", + "domain", + "from", + "from_name", + "to", + "cc", + "subject", + "received_at", + "size", + "attachment_count", + "verdicts", + "truncated", + "parse_failed", + "raw_available", + "text", + "html", + "envelope_from", + "recipients", + "reply_to", + "message_id", + "in_reply_to", + "references", + "attachments", + "headers" + ], + "title": "InboundResponse", + "type": "object" + }, + "InboundSummary": { + "description": "A received message, without its body. What the list returns.", + "properties": { + "attachment_count": { + "description": "How many attachments it has.", + "examples": [ + 0 + ], + "title": "Attachment Count", + "type": "integer" + }, + "cc": { + "description": "The addresses in the `Cc` header. Empty if none.", + "items": { + "type": "string" + }, + "title": "Cc", + "type": "array" + }, + "domain": { + "description": "The domain the message was received for.", + "examples": [ + "example.com" + ], + "title": "Domain", + "type": "string" + }, + "from": { + "description": "The address in the `From` header, lower-cased. Empty if there was none. Anybody can write anything here: see `verdicts` for what SES checked.", + "examples": [ + "ada@other.org" + ], + "title": "From", + "type": "string" + }, + "from_name": { + "description": "The display name in `From`, if any.", + "examples": [ + "Ada" + ], + "title": "From Name", + "type": "string" + }, + "id": { + "description": "Opaque and stable, prefixed `inbound_`.", + "examples": [ + "inbound_01J8XQ2K3M4N5P6Q7R8S9T0V1W" + ], + "title": "Id", + "type": "string" + }, + "parse_failed": { + "description": "Whether the message could not be read at all. Everything else is empty; the original is available through `/raw` until it expires.", + "title": "Parse Failed", + "type": "boolean" + }, + "raw_available": { + "description": "Whether the original is expected to still be in storage. Retention removes it at a day boundary, so a message just past `raw_expires_at` may already be gone and one just before it should not be.", + "title": "Raw Available", + "type": "boolean" + }, + "raw_expires_at": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "description": "When retention will have removed the original and its attachments. Null when unknown. The parsed message below stays regardless. UTC.", + "title": "Raw Expires At" + }, + "received_at": { + "description": "When SES received the message. UTC.", + "format": "date-time", + "title": "Received At", + "type": "string" + }, + "size": { + "description": "Size of the whole message in bytes.", + "examples": [ + 4821 + ], + "title": "Size", + "type": "integer" + }, + "subject": { + "description": "The subject, decoded and on one line.", + "title": "Subject", + "type": "string" + }, + "to": { + "description": "The addresses in the `To` header.", + "items": { + "type": "string" + }, + "title": "To", + "type": "array" + }, + "truncated": { + "description": "Whether a limit was hit while reading it, so something is missing from this representation. The original has all of it.", + "title": "Truncated", + "type": "boolean" + }, + "verdicts": { + "$ref": "#/components/schemas/InboundVerdicts", + "description": "What SES concluded about the message." + } + }, + "required": [ + "id", + "domain", + "from", + "from_name", + "to", + "cc", + "subject", + "received_at", + "size", + "attachment_count", + "verdicts", + "truncated", + "parse_failed", + "raw_available" + ], + "title": "InboundSummary", + "type": "object" + }, + "InboundVerdicts": { + "description": "What SES concluded about the message. Recorded, never acted on.", + "properties": { + "dkim": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "description": "The DKIM verdict, in the same values as `spf`. `GRAY` means the message was not signed or the signing domain does not match the sender's.", + "examples": [ + "GRAY" + ], + "title": "Dkim" + }, + "dmarc": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "description": "The DMARC verdict, in the same values as `spf`.", + "title": "Dmarc" + }, + "dmarc_policy": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "description": "The sender domain's published policy - `none`, `quarantine` or `reject` - and present only when DMARC failed.", + "title": "Dmarc Policy" + }, + "spam": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "description": "The spam verdict, in the same values.", + "title": "Spam" + }, + "spf": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "description": "`PASS`, `FAIL`, `GRAY` or `PROCESSING_FAILED`. Null means SES did not say, which is not the same as `PASS`.", + "examples": [ + "PASS" + ], + "title": "Spf" + }, + "virus": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "description": "The virus verdict, in the same values.", + "title": "Virus" + } + }, + "title": "InboundVerdicts", + "type": "object" + }, "ReadinessResponse": { "properties": { "dependencies": { @@ -1054,61 +1591,277 @@ "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/HealthResponse" + "$ref": "#/components/schemas/HealthResponse" + } + } + }, + "description": "Successful Response" + } + }, + "summary": "Liveness probe", + "tags": [ + "health" + ] + } + }, + "/readyz": { + "get": { + "description": "Round-trip both Postgres and Redis.\n\nReturns 503 if either is unreachable, so a partially-started stack is not\nadvertised as ready.", + "operationId": "readyz_readyz_get", + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ReadinessResponse" + } + } + }, + "description": "Successful Response" + }, + "503": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ReadinessResponse" + } + } + }, + "description": "A dependency is unavailable" + } + }, + "summary": "Readiness probe", + "tags": [ + "health" + ] + } + }, + "/v1/api-keys": { + "get": { + "description": "Every key belonging to the calling key's project.\n\nScoped by the authenticated project rather than by anything in the request,\nso there is no parameter a caller could change to see someone else's keys.", + "operationId": "list_keys_v1_api_keys_get", + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/APIKeyList" + } + } + }, + "description": "Successful Response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + }, + "description": "Invalid or missing API key." + }, + "429": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + }, + "description": "Rate limit exceeded." + } + }, + "summary": "List API keys", + "tags": [ + "api-keys" + ] + } + }, + "/v1/domains": { + "get": { + "description": "Every domain this key's project can send from.", + "operationId": "list_domains_v1_domains_get", + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/DomainList" + } + } + }, + "description": "Successful Response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + }, + "description": "Invalid or missing API key." + }, + "429": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + }, + "description": "Rate limit exceeded." + } + }, + "summary": "List sending domains", + "tags": [ + "domains" + ] + } + }, + "/v1/emails": { + "get": { + "description": "This key's project, newest first.\n\n**Paged by cursor rather than by offset.** Ids sort in the order they were\ncreated, so ``starting_after`` names a fixed point in the list and stays\ncorrect while new messages arrive underneath it. An offset does not: a send\nbetween two pages shifts every row down one, and the reader silently skips\nthe message that moved across the boundary. For a send log that is data\nloss nobody can see.\n\nAn unknown ``starting_after`` is a 404 rather than an empty page. The\ncomparison is lexical, so an id from another project would otherwise return\na page of real messages positioned by an id the caller cannot see - a wrong\nanswer that looks like a right one.", + "operationId": "list_emails_v1_emails_get", + "parameters": [ + { + "description": "How many messages to return, newest first.", + "in": "query", + "name": "limit", + "required": false, + "schema": { + "default": 25, + "description": "How many messages to return, newest first.", + "maximum": 100, + "minimum": 1, + "title": "Limit", + "type": "integer" + } + }, + { + "description": "Return messages older than this id - the last id from the previous page. The id must belong to this project.", + "in": "query", + "name": "starting_after", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "description": "Return messages older than this id - the last id from the previous page. The id must belong to this project.", + "title": "Starting After" + } + }, + { + "description": "Only messages in this status.", + "in": "query", + "name": "status", + "required": false, + "schema": { + "anyOf": [ + { + "$ref": "#/components/schemas/EmailStatus" + }, + { + "type": "null" + } + ], + "description": "Only messages in this status.", + "title": "Status" + } + } + ], + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/EmailList" + } + } + }, + "description": "Successful Response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" } } }, - "description": "Successful Response" - } - }, - "summary": "Liveness probe", - "tags": [ - "health" - ] - } - }, - "/readyz": { - "get": { - "description": "Round-trip both Postgres and Redis.\n\nReturns 503 if either is unreachable, so a partially-started stack is not\nadvertised as ready.", - "operationId": "readyz_readyz_get", - "responses": { - "200": { + "description": "Invalid or missing API key." + }, + "422": { "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/ReadinessResponse" + "$ref": "#/components/schemas/HTTPValidationError" } } }, - "description": "Successful Response" + "description": "Validation Error" }, - "503": { + "429": { "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/ReadinessResponse" + "$ref": "#/components/schemas/ErrorResponse" } } }, - "description": "A dependency is unavailable" + "description": "Rate limit exceeded." } }, - "summary": "Readiness probe", + "summary": "List emails", "tags": [ - "health" + "emails" ] - } - }, - "/v1/api-keys": { - "get": { - "description": "Every key belonging to the calling key's project.\n\nScoped by the authenticated project rather than by anything in the request,\nso there is no parameter a caller could change to see someone else's keys.", - "operationId": "list_keys_v1_api_keys_get", + }, + "post": { + "description": "Accept a message for sending.", + "operationId": "send_email_v1_emails_post", + "parameters": [ + { + "description": "Repeat a request safely. A second send with the same key returns the first message's id and sends nothing further, so a retry after a timeout cannot deliver twice. Scoped to the project.", + "in": "header", + "name": "Idempotency-Key", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "description": "Repeat a request safely. A second send with the same key returns the first message's id and sends nothing further, so a retry after a timeout cannot deliver twice. Scoped to the project.", + "title": "Idempotency-Key" + } + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SendEmailRequest" + } + } + }, + "required": true + }, "responses": { - "200": { + "201": { "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/APIKeyList" + "$ref": "#/components/schemas/SendEmailResponse" } } }, @@ -1124,6 +1877,16 @@ }, "description": "Invalid or missing API key." }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + }, "429": { "content": { "application/json": { @@ -1135,22 +1898,35 @@ "description": "Rate limit exceeded." } }, - "summary": "List API keys", + "summary": "Send an email", "tags": [ - "api-keys" + "emails" ] } }, - "/v1/domains": { + "/v1/emails/{email_id}": { "get": { - "description": "Every domain this key's project can send from.", - "operationId": "list_domains_v1_domains_get", + "description": "One message, if it belongs to this key's project.\n\nOwnership is part of the query, so an id from another project is a 404\nrather than a 403 - which would confirm the id exists.", + "operationId": "get_email_v1_emails__email_id__get", + "parameters": [ + { + "description": "The id returned when the message was accepted.", + "in": "path", + "name": "email_id", + "required": true, + "schema": { + "description": "The id returned when the message was accepted.", + "title": "Email Id", + "type": "string" + } + } + ], "responses": { "200": { "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/DomainList" + "$ref": "#/components/schemas/EmailResponse" } } }, @@ -1166,6 +1942,16 @@ }, "description": "Invalid or missing API key." }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + }, "429": { "content": { "application/json": { @@ -1177,16 +1963,16 @@ "description": "Rate limit exceeded." } }, - "summary": "List sending domains", + "summary": "Retrieve an email", "tags": [ - "domains" + "emails" ] } }, - "/v1/emails": { + "/v1/inbound": { "get": { - "description": "This key's project, newest first.\n\n**Paged by cursor rather than by offset.** Ids sort in the order they were\ncreated, so ``starting_after`` names a fixed point in the list and stays\ncorrect while new messages arrive underneath it. An offset does not: a send\nbetween two pages shifts every row down one, and the reader silently skips\nthe message that moved across the boundary. For a send log that is data\nloss nobody can see.\n\nAn unknown ``starting_after`` is a 404 rather than an empty page. The\ncomparison is lexical, so an id from another project would otherwise return\na page of real messages positioned by an id the caller cannot see - a wrong\nanswer that looks like a right one.", - "operationId": "list_emails_v1_emails_get", + "description": "This key's project, newest first, without bodies.\n\nPaged by cursor for the reason ``/v1/emails`` is: ids sort in the order they\nwere created, so ``starting_after`` names a fixed point and stays correct\nwhile mail arrives underneath it, where an offset would silently skip the\nmessage that moved across the boundary. An unknown ``starting_after`` is a 404\nand not an empty page, because the comparison is lexical and an id from\nanother project would otherwise position a page of real messages by a value\nthe caller cannot see.\n\nSorted by id and not by ``received_at``. A backlog means a message can be\nrecorded long after it arrived, so the two orders can differ; id order is the\none a cursor can be stable over, and ``received_at`` is on every row.", + "operationId": "list_inbound_v1_inbound_get", "parameters": [ { "description": "How many messages to return, newest first.", @@ -1221,21 +2007,21 @@ } }, { - "description": "Only messages in this status.", + "description": "Only messages received for this domain.", "in": "query", - "name": "status", + "name": "domain", "required": false, "schema": { "anyOf": [ { - "$ref": "#/components/schemas/EmailStatus" + "type": "string" }, { "type": "null" } ], - "description": "Only messages in this status.", - "title": "Status" + "description": "Only messages received for this domain.", + "title": "Domain" } } ], @@ -1244,7 +2030,7 @@ "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/EmailList" + "$ref": "#/components/schemas/InboundList" } } }, @@ -1281,50 +2067,35 @@ "description": "Rate limit exceeded." } }, - "summary": "List emails", + "summary": "List received emails", "tags": [ - "emails" + "inbound" ] - }, - "post": { - "description": "Accept a message for sending.", - "operationId": "send_email_v1_emails_post", + } + }, + "/v1/inbound/{inbound_id}": { + "get": { + "description": "One message, parsed. Bodies, headers, attachments and what SES concluded.\n\nThe bodies are the sender's, not SESKit's. In particular the HTML is whatever\nwas sent, and has to be treated as hostile by anything that renders it.", + "operationId": "get_inbound_v1_inbound__inbound_id__get", "parameters": [ { - "description": "Repeat a request safely. A second send with the same key returns the first message's id and sends nothing further, so a retry after a timeout cannot deliver twice. Scoped to the project.", - "in": "header", - "name": "Idempotency-Key", - "required": false, + "description": "The id from the list, or from an `email.received` event.", + "in": "path", + "name": "inbound_id", + "required": true, "schema": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "description": "Repeat a request safely. A second send with the same key returns the first message's id and sends nothing further, so a retry after a timeout cannot deliver twice. Scoped to the project.", - "title": "Idempotency-Key" + "description": "The id from the list, or from an `email.received` event.", + "title": "Inbound Id", + "type": "string" } } ], - "requestBody": { - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/SendEmailRequest" - } - } - }, - "required": true - }, "responses": { - "201": { + "200": { "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/SendEmailResponse" + "$ref": "#/components/schemas/InboundResponse" } } }, @@ -1361,39 +2132,118 @@ "description": "Rate limit exceeded." } }, - "summary": "Send an email", + "summary": "Retrieve a received email", "tags": [ - "emails" + "inbound" ] } }, - "/v1/emails/{email_id}": { + "/v1/inbound/{inbound_id}/attachments/{index}": { "get": { - "description": "One message, if it belongs to this key's project.\n\nOwnership is part of the query, so an id from another project is a 404\nrather than a 403 - which would confirm the id exists.", - "operationId": "get_email_v1_emails__email_id__get", + "description": "One attachment's bytes, fetched from the stored message.\n\n**Always served as `application/octet-stream`**, whatever the sender declared,\nwith `Content-Disposition: attachment`, `nosniff` and a sandboxing CSP. The\ndeclared type is on the message's `attachments` for you to read; it is never\none this endpoint believes. These are files a stranger sent, served from your\nAPI's origin.\n\nNeeds the stored message, so it stops working when retention removes it. The\nparsed message does not.", + "operationId": "download_attachment_v1_inbound__inbound_id__attachments__index__get", "parameters": [ { - "description": "The id returned when the message was accepted.", + "description": "The id of the received message.", "in": "path", - "name": "email_id", + "name": "inbound_id", "required": true, "schema": { - "description": "The id returned when the message was accepted.", - "title": "Email Id", + "description": "The id of the received message.", + "title": "Inbound Id", "type": "string" } + }, + { + "description": "The attachment's `index` from the message's `attachments`.", + "in": "path", + "name": "index", + "required": true, + "schema": { + "description": "The attachment's `index` from the message's `attachments`.", + "minimum": 0, + "title": "Index", + "type": "integer" + } } ], "responses": { "200": { + "content": { + "application/octet-stream": { + "schema": { + "format": "binary", + "type": "string" + } + } + }, + "description": "The bytes." + }, + "401": { "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/EmailResponse" + "$ref": "#/components/schemas/ErrorResponse" } } }, - "description": "Successful Response" + "description": "Invalid or missing API key." + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + }, + "429": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + }, + "description": "Rate limit exceeded." + } + }, + "summary": "Download an attachment", + "tags": [ + "inbound" + ] + } + }, + "/v1/inbound/{inbound_id}/raw": { + "get": { + "description": "The message exactly as SES received it, as an `.eml` file.\n\nUnparsed and unmodified: every header, every part, every attachment. Useful for\ndebugging a message that did not parse as expected, and for handing to another\ntool. It is the sender's bytes, served as a download with the same protections\nas an attachment.\n\nNeeds the stored message, so it stops working when retention removes it.", + "operationId": "download_raw_v1_inbound__inbound_id__raw_get", + "parameters": [ + { + "description": "The id of the received message.", + "in": "path", + "name": "inbound_id", + "required": true, + "schema": { + "description": "The id of the received message.", + "title": "Inbound Id", + "type": "string" + } + } + ], + "responses": { + "200": { + "content": { + "message/rfc822": { + "schema": { + "format": "binary", + "type": "string" + } + } + }, + "description": "The original message, byte for byte." }, "401": { "content": { @@ -1426,9 +2276,9 @@ "description": "Rate limit exceeded." } }, - "summary": "Retrieve an email", + "summary": "Download the original message", "tags": [ - "emails" + "inbound" ] } }, diff --git a/docs/reference/sdk.md b/docs/reference/sdk.md index fa0640a..a1f6d5e 100644 --- a/docs/reference/sdk.md +++ b/docs/reference/sdk.md @@ -107,6 +107,33 @@ while True: cursor = page.last_id ``` +## Received mail + +```python +page = client.inbound.list(domain="example.com") +message = client.inbound.get(page.data[0].id) + +print(message.from_, message.subject) +print(message.text) + +pdf = client.inbound.attachment(message.id, 0) # bytes +eml = client.inbound.raw(message.id) # bytes +``` + +`list` returns **summaries** — who, what, when, how big, and what Amazon SES +concluded — with no bodies, so a page of a hundred stays small. `get` returns +the whole message. Like `emails.list`, it is **one page**: pass `page.last_id` +as `starting_after` while `page.has_more`. + +`attachment` and `raw` return **bytes, untouched**. They need the stored +original, so they raise `NotFound` once your [retention period](configuration.md) +has removed it. The parsed message from `get` does not expire. + +**Everything in a received message was written by somebody else.** `html` in +particular is exactly what was sent and may contain script, tracking images and +links that lie; do not render it in a page that carries your own credentials. +A verdict that is `None` means Amazon SES did not say — it is not a pass. + ## Async The same surface, awaited: diff --git a/migrations/versions/c4e81a7d5b29_events_for_received_mail.py b/migrations/versions/c4e81a7d5b29_events_for_received_mail.py new file mode 100644 index 0000000..a7d8c19 --- /dev/null +++ b/migrations/versions/c4e81a7d5b29_events_for_received_mail.py @@ -0,0 +1,74 @@ +"""events for received mail + +An event was always about a message SESKit sent, so `email_events.email_id` was +required. A message that *arrives* has no `emails` row, and the webhook +machinery - signing, retries, auto-disable, the deliveries log - is all keyed to +an event. Giving received mail its own events and deliveries would mean writing +that twice, so this lets an event belong to either instead. + +`email_id` becomes nullable and `inbound_email_id` is added beside it. A check +constraint holds that exactly one is set: neither would leave an event nothing +can be delivered for, since the project is found through the parent, and both +would make it ambiguous which. + +Every existing row keeps its `email_id` and gets a NULL `inbound_email_id`, which +is the only shape that satisfies the constraint, so nothing is rewritten. The +constraint is added `NOT VALID` and then validated separately: validation scans +the table but takes a weaker lock than adding it checked would, and this table +grows with every message anyone sends. + +**Downgrade deletes received events.** They have no `email_id` to put back, and +reverting the column to NOT NULL would otherwise fail on them. Their webhook +deliveries go with them. Received messages themselves are untouched. + +Revision ID: c4e81a7d5b29 +Revises: b6d94e0a31c7 +Create Date: 2026-10-05 15:02:11.719304 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "c4e81a7d5b29" +down_revision: str | Sequence[str] | None = "b6d94e0a31c7" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +ONE_PARENT = ( + "(email_id IS NOT NULL AND inbound_email_id IS NULL) " + "OR (email_id IS NULL AND inbound_email_id IS NOT NULL)" +) + + +def upgrade() -> None: + op.alter_column("email_events", "email_id", existing_type=sa.String(length=64), nullable=True) + op.add_column( + "email_events", + sa.Column( + "inbound_email_id", + sa.String(length=64), + sa.ForeignKey("inbound_emails.id", ondelete="CASCADE"), + nullable=True, + ), + ) + op.create_index( + op.f("ix_email_events_inbound_email_id"), + "email_events", + ["inbound_email_id"], + unique=False, + ) + op.execute( + f"ALTER TABLE email_events ADD CONSTRAINT ck_email_events_one_parent " + f"CHECK ({ONE_PARENT}) NOT VALID" + ) + op.execute("ALTER TABLE email_events VALIDATE CONSTRAINT ck_email_events_one_parent") + + +def downgrade() -> None: + op.execute("DELETE FROM email_events WHERE email_id IS NULL") + op.execute("ALTER TABLE email_events DROP CONSTRAINT ck_email_events_one_parent") + op.drop_index(op.f("ix_email_events_inbound_email_id"), table_name="email_events") + op.drop_column("email_events", "inbound_email_id") + op.alter_column("email_events", "email_id", existing_type=sa.String(length=64), nullable=False) diff --git a/packages/core/src/seskit_core/events/__init__.py b/packages/core/src/seskit_core/events/__init__.py index 50c78b6..6b4bc22 100644 --- a/packages/core/src/seskit_core/events/__init__.py +++ b/packages/core/src/seskit_core/events/__init__.py @@ -18,6 +18,7 @@ NotAReceipt, ReceivedNotification, candidate_domains, + event_for, is_recorded, locate, parse_received, @@ -34,6 +35,7 @@ summarise, suppression_reason, to_public, + to_public_received, ) from seskit_core.events.origin import TopicOrigin, connections_for_origin, parse_topic_arn @@ -51,6 +53,7 @@ "apply_to_email", "candidate_domains", "connections_for_origin", + "event_for", "event_name", "ingest_event", "is_recorded", @@ -66,5 +69,6 @@ "summarise", "suppression_reason", "to_public", + "to_public_received", "unwrap", ] diff --git a/packages/core/src/seskit_core/events/inbound.py b/packages/core/src/seskit_core/events/inbound.py index 935167e..8eced4f 100644 --- a/packages/core/src/seskit_core/events/inbound.py +++ b/packages/core/src/seskit_core/events/inbound.py @@ -39,8 +39,11 @@ from seskit_core.email.parse import parse_message from seskit_core.events.ingest import Outcome +from seskit_core.events.normalise import to_public_received +from seskit_core.ids import IDPrefix, generate_id from seskit_core.logging import get_logger -from seskit_core.models import Identity, InboundEmail +from seskit_core.models import EmailEvent, EventType, Identity, InboundEmail +from seskit_core.services.webhooks import queue_deliveries logger = get_logger(__name__) @@ -317,6 +320,11 @@ async def record_received( ) return Outcome.DUPLICATE, existing + # In the same transaction as the row, so a webhook cannot be lost to the + # process dying between the two. A duplicate returned above and never gets + # here, so it raises no second event. + await _raise_received(session, row) + # Ids and sizes only. This is somebody's mail; what is in it, and who it is # from and to, is not a reason to put any of it in a log. logger.info( @@ -328,3 +336,82 @@ async def record_received( parse_failed=row.parse_failed, ) return Outcome.RECORDED, row + + +def received_data(row: InboundEmail) -> dict[str, Any]: + """What a webhook says about a message that arrived. + + A description, not the message. No body, no headers and no attachment bytes: + the payload stays small however large the mail is, and a receiver that wants + more asks for it with its API key, which is what the endpoint is for. The + subject and sender are here because a receiver deciding whether to bother + fetching needs them, and because they are the same fields every other mail + webhook leads with. + + ``to`` is the header; ``recipients`` is who the message was actually + addressed to on the wire, which is the address that received it and can be + one the headers never mention. + """ + return { + "domain": row.domain, + "from": row.from_address, + "from_name": row.from_name, + "to": row.to_addresses, + "cc": row.cc_addresses, + "recipients": row.envelope_to, + "subject": row.subject, + "received_at": row.received_at.isoformat(), + "size": row.size_bytes, + "attachment_count": len(row.attachments), + # Verbatim, and null where SES did not say - which is not a PASS. + "verdicts": { + "spf": row.spf_verdict, + "dkim": row.dkim_verdict, + "dmarc": row.dmarc_verdict, + "spam": row.spam_verdict, + "virus": row.virus_verdict, + }, + "truncated": row.truncated, + "parse_failed": row.parse_failed, + } + + +async def _raise_received(session: AsyncSession, row: InboundEmail) -> EmailEvent: + """Record that mail arrived, and queue it for every endpoint that wants it. + + ``occurred_at`` is when SES received the message, not when this ran: a + backlog means SESKit can hear about mail long after it arrived, and + ``created_at`` on the payload says when it *happened*. + + ``provider_event_id`` is null. The unique column is the SNS id for events + SES sent about mail SESKit sent; deduplication for received mail is on the + row itself, which is unique on its storage location and its own SNS id. + """ + event_id = generate_id(IDPrefix.EVENT) + event = EmailEvent( + id=event_id, + inbound_email_id=row.id, + event_type=EventType.RECEIVED.value, + provider_event_id=None, + occurred_at=row.received_at, + payload=to_public_received( + event_id=event_id, + inbound_id=row.id, + occurred=row.received_at, + data=received_data(row), + ), + ) + session.add(event) + await session.flush() + await queue_deliveries(session, event) + return event + + +async def event_for(session: AsyncSession, inbound_id: str) -> EmailEvent | None: + """The ``received`` event raised for a message, so a caller can ask the queue + to attempt its deliveries now instead of waiting for the sweep. + """ + found: EmailEvent | None = await session.scalar( + select(EmailEvent).where(EmailEvent.inbound_email_id == inbound_id) + ) + return found diff --git a/packages/core/src/seskit_core/events/normalise.py b/packages/core/src/seskit_core/events/normalise.py index 9b98191..085b0db 100644 --- a/packages/core/src/seskit_core/events/normalise.py +++ b/packages/core/src/seskit_core/events/normalise.py @@ -226,3 +226,26 @@ def to_public( "created_at": occurred.isoformat(), "data": data, } + + +def to_public_received( + *, + event_id: str, + inbound_id: str, + occurred: datetime, + data: dict[str, Any], +) -> dict[str, Any]: + """The normalised event for a message that arrived. + + The same envelope as :func:`to_public` with ``inbound_id`` where ``email_id`` + would be. Not an ``email_id`` that happens to hold a different kind of id: a + receiver switching on the field's presence has to be able to trust what it + refers to, and ``GET /v1/emails/{id}`` must never be handed an inbound id. + """ + return { + "id": event_id, + "type": f"email.{EventType.RECEIVED.value}", + "inbound_id": inbound_id, + "created_at": occurred.isoformat(), + "data": data, + } diff --git a/packages/core/src/seskit_core/models/email_event.py b/packages/core/src/seskit_core/models/email_event.py index 9ff47a9..f1163b1 100644 --- a/packages/core/src/seskit_core/models/email_event.py +++ b/packages/core/src/seskit_core/models/email_event.py @@ -17,7 +17,7 @@ from enum import StrEnum from typing import TYPE_CHECKING -from sqlalchemy import JSON, DateTime, ForeignKey, Index, String +from sqlalchemy import JSON, CheckConstraint, DateTime, ForeignKey, Index, String from sqlalchemy.orm import Mapped, mapped_column, relationship from seskit_core.db import Base @@ -26,6 +26,7 @@ if TYPE_CHECKING: from seskit_core.models.email import Email + from seskit_core.models.inbound_email import InboundEmail class EventType(StrEnum): @@ -50,6 +51,11 @@ class EventType(StrEnum): #: bounces and reimplementing the permanent/transient rule. SUPPRESSED = "suppressed" + #: Mail arrived for one of the project's domains. The only type that is not + #: about a message SESKit sent: its parent is an ``InboundEmail``, not an + #: ``Email``, and it is the one event with no ``email_id``. + RECEIVED = "received" + REJECTED = "rejected" DELIVERY_DELAYED = "delivery_delayed" RENDERING_FAILED = "rendering_failed" @@ -71,6 +77,7 @@ class EventType(StrEnum): EventType.OPENED, EventType.CLICKED, EventType.SUPPRESSED, + EventType.RECEIVED, } ) @@ -85,6 +92,7 @@ class EventType(StrEnum): EventType.OPENED: "Opened", EventType.CLICKED: "Link clicked", EventType.SUPPRESSED: "Added to the suppression list", + EventType.RECEIVED: "Received", EventType.REJECTED: "Rejected by SES", EventType.DELIVERY_DELAYED: "Delivery delayed", EventType.RENDERING_FAILED: "Template rendering failed", @@ -101,6 +109,16 @@ class EmailEvent(Base, TimestampMixin): # Newest-first for one message, which is how both the detail page and # the API read them. Index("ix_email_events_email_occurred", "email_id", "occurred_at"), + # An event is about exactly one thing: a message SESKit sent, or a + # message that arrived. Neither would leave an event nothing can be + # delivered for - the project is found through the parent - and both + # would make it ambiguous which. The database holds this, so no path + # through the code can break it. + CheckConstraint( + "(email_id IS NOT NULL AND inbound_email_id IS NULL) " + "OR (email_id IS NULL AND inbound_email_id IS NOT NULL)", + name="ck_email_events_one_parent", + ), ) id: Mapped[str] = mapped_column( @@ -109,11 +127,21 @@ class EmailEvent(Base, TimestampMixin): default=lambda: generate_id(IDPrefix.EVENT), ) - email_id: Mapped[str] = mapped_column( + #: The message this is about, when SESKit sent it. NULL for `received`, + #: which is about ``inbound_email_id`` instead - see the check constraint. + email_id: Mapped[str | None] = mapped_column( String(64), ForeignKey("emails.id", ondelete="CASCADE"), index=True, - nullable=False, + nullable=True, + ) + + #: The message this is about, when it arrived. + inbound_email_id: Mapped[str | None] = mapped_column( + String(64), + ForeignKey("inbound_emails.id", ondelete="CASCADE"), + index=True, + nullable=True, ) event_type: Mapped[str] = mapped_column(String(32), nullable=False, index=True) @@ -148,7 +176,8 @@ class EmailEvent(Base, TimestampMixin): DateTime(timezone=True), nullable=False, index=True ) - email: Mapped[Email] = relationship(back_populates="events") + email: Mapped[Email | None] = relationship(back_populates="events") + inbound_email: Mapped[InboundEmail | None] = relationship() @property def type(self) -> EventType: diff --git a/packages/core/src/seskit_core/services/__init__.py b/packages/core/src/seskit_core/services/__init__.py index 5837fcb..f3557ef 100644 --- a/packages/core/src/seskit_core/services/__init__.py +++ b/packages/core/src/seskit_core/services/__init__.py @@ -58,6 +58,7 @@ ) from seskit_core.services.inbound import ( InboundProvisionerFactory, + InboundStoreFactory, PolledInbox, bucket_name_for, count_other_receivers, @@ -136,6 +137,7 @@ "ActivityPoint", "EmailAlreadyRegistered", "InboundProvisionerFactory", + "InboundStoreFactory", "IssuedKey", "Metrics", "Outgoing", diff --git a/packages/core/src/seskit_core/services/inbound.py b/packages/core/src/seskit_core/services/inbound.py index 1e6d677..9e67057 100644 --- a/packages/core/src/seskit_core/services/inbound.py +++ b/packages/core/src/seskit_core/services/inbound.py @@ -47,6 +47,7 @@ InboundInfrastructure, InboundProvisioner, InboundRule, + InboundStore, ) from seskit_core.services.credentials import stored_credentials @@ -56,6 +57,10 @@ #: ``ProvisionerFactory``: so this module never imports an adapter. InboundProvisionerFactory = Callable[[str, AWSCredentials], InboundProvisioner] +#: Builds a reader of stored messages for a region, on one project's credentials. +#: Injected so a route never imports an adapter, and a test can substitute one. +InboundStoreFactory = Callable[[str, AWSCredentials], InboundStore] + #: Every rule name starts with this. It must equal the adapter's #: ``RULE_NAME_PREFIX``, because the bucket policy admits only rules that carry #: it - a rule named otherwise would be accepted by SES and its writes refused, diff --git a/packages/core/src/seskit_core/services/webhooks.py b/packages/core/src/seskit_core/services/webhooks.py index 3fdb569..010a99c 100644 --- a/packages/core/src/seskit_core/services/webhooks.py +++ b/packages/core/src/seskit_core/services/webhooks.py @@ -33,6 +33,7 @@ DeliveryStatus, Email, EmailEvent, + InboundEmail, WebhookDelivery, WebhookEndpoint, WebhookStatus, @@ -232,6 +233,24 @@ async def delete_endpoint(session: AsyncSession, endpoint: WebhookEndpoint) -> N # -------------------------------------------------------------- queueing --- +async def _project_of(session: AsyncSession, event: EmailEvent) -> str | None: + """The project an event belongs to, found through whichever parent it has. + + A message SESKit sent, or one that arrived. An event has exactly one - the + database holds that - so exactly one of these runs, and an event with + neither (which cannot be stored) delivers to nobody. + """ + if event.email_id is not None: + found = await session.scalar(select(Email.project_id).where(Email.id == event.email_id)) + elif event.inbound_email_id is not None: + found = await session.scalar( + select(InboundEmail.project_id).where(InboundEmail.id == event.inbound_email_id) + ) + else: + return None + return str(found) if found is not None else None + + async def queue_deliveries(session: AsyncSession, event: EmailEvent) -> list[WebhookDelivery]: """Create a delivery row per enabled endpoint on the event's project. @@ -245,7 +264,7 @@ async def queue_deliveries(session: AsyncSession, event: EmailEvent) -> list[Web if event.type not in PUBLIC_EVENT_TYPES: return [] - project_id = await session.scalar(select(Email.project_id).where(Email.id == event.email_id)) + project_id = await _project_of(session, event) if project_id is None: return [] diff --git a/packages/sdk-python/src/seskit/__init__.py b/packages/sdk-python/src/seskit/__init__.py index cd741cc..f911283 100644 --- a/packages/sdk-python/src/seskit/__init__.py +++ b/packages/sdk-python/src/seskit/__init__.py @@ -29,8 +29,8 @@ first-class way to use SESKit. """ -from seskit._async_client import AsyncEmails, AsyncSesKit -from seskit._client import Emails, SesKit +from seskit._async_client import AsyncEmails, AsyncInbound, AsyncSesKit +from seskit._client import Emails, Inbound, SesKit from seskit._errors import ( AttachmentTooLarge, AuthenticationFailed, @@ -48,7 +48,16 @@ SESKitError, SuppressedRecipient, ) -from seskit._models import Accepted, Email, EmailPage +from seskit._models import ( + Accepted, + Email, + EmailPage, + InboundAttachment, + InboundEmail, + InboundPage, + InboundSummary, + Verdicts, +) from seskit.resources import Attachment __version__ = "0.1.0" @@ -56,6 +65,7 @@ __all__ = [ "Accepted", "AsyncEmails", + "AsyncInbound", "AsyncSesKit", "Attachment", "AttachmentTooLarge", @@ -66,6 +76,11 @@ "EmailPage", "EmailRejected", "Emails", + "Inbound", + "InboundAttachment", + "InboundEmail", + "InboundPage", + "InboundSummary", "InternalError", "InvalidRecipient", "InvalidRequest", @@ -77,5 +92,6 @@ "SendingLimitExceeded", "SesKit", "SuppressedRecipient", + "Verdicts", "__version__", ] diff --git a/packages/sdk-python/src/seskit/_async_client.py b/packages/sdk-python/src/seskit/_async_client.py index 2468f65..b750916 100644 --- a/packages/sdk-python/src/seskit/_async_client.py +++ b/packages/sdk-python/src/seskit/_async_client.py @@ -26,14 +26,23 @@ import httpx -from seskit._models import Accepted, Email, EmailPage +from seskit._models import Accepted, Email, EmailPage, InboundEmail, InboundPage from seskit._transport import ( DEFAULT_MAX_ATTEMPTS, DEFAULT_TIMEOUT, BaseTransport, Request, ) -from seskit.resources import Attachment, build_get, build_list, build_send +from seskit.resources import ( + Attachment, + build_get, + build_inbound_attachment, + build_inbound_get, + build_inbound_list, + build_inbound_raw, + build_list, + build_send, +) class _AsyncTransport(BaseTransport): @@ -66,11 +75,11 @@ async def send(self, request: Request) -> Any: continue if not self.should_retry(status_code=last.status_code, attempt=attempt + 1): - return self.unwrap(last) + return self.unwrap(last, binary=request.binary) await asyncio.sleep(self.backoff(last, attempt)) assert last is not None # noqa: S101 - unreachable; the loop ran at least once - return self.unwrap(last) + return self.unwrap(last, binary=request.binary) async def aclose(self) -> None: if self._owns_client: @@ -134,6 +143,40 @@ async def list( return EmailPage.from_payload(payload) +class AsyncInbound: + """`client.inbound` - mail your project received.""" + + def __init__(self, transport: _AsyncTransport) -> None: + self._transport = transport + + async def get(self, inbound_id: str) -> InboundEmail: + """One received message, parsed. `NotFound` if it is not this key's project's.""" + return InboundEmail.from_payload(await self._transport.send(build_inbound_get(inbound_id))) + + async def list( + self, + *, + limit: int | None = None, + starting_after: str | None = None, + domain: str | None = None, + ) -> InboundPage: + """One page of received messages, newest first, without bodies.""" + payload = await self._transport.send( + build_inbound_list(limit=limit, starting_after=starting_after, domain=domain) + ) + return InboundPage.from_payload(payload) + + async def attachment(self, inbound_id: str, index: int) -> bytes: + """One attachment's bytes. `index` is from the message's `attachments`.""" + data: bytes = await self._transport.send(build_inbound_attachment(inbound_id, index)) + return data + + async def raw(self, inbound_id: str) -> bytes: + """The message exactly as it was received, as the bytes of an `.eml`.""" + data: bytes = await self._transport.send(build_inbound_raw(inbound_id)) + return data + + class AsyncSesKit: """A SESKit instance, addressed over HTTP without blocking the loop. @@ -162,6 +205,7 @@ def __init__( client=http_client, ) self.emails = AsyncEmails(self._transport) + self.inbound = AsyncInbound(self._transport) async def aclose(self) -> None: """Release the connection pool. Not needed for a client you kept.""" diff --git a/packages/sdk-python/src/seskit/_client.py b/packages/sdk-python/src/seskit/_client.py index 1a70015..89945e1 100644 --- a/packages/sdk-python/src/seskit/_client.py +++ b/packages/sdk-python/src/seskit/_client.py @@ -20,14 +20,23 @@ import httpx -from seskit._models import Accepted, Email, EmailPage +from seskit._models import Accepted, Email, EmailPage, InboundEmail, InboundPage from seskit._transport import ( DEFAULT_MAX_ATTEMPTS, DEFAULT_TIMEOUT, BaseTransport, Request, ) -from seskit.resources import Attachment, build_get, build_list, build_send +from seskit.resources import ( + Attachment, + build_get, + build_inbound_attachment, + build_inbound_get, + build_inbound_list, + build_inbound_raw, + build_list, + build_send, +) class _Transport(BaseTransport): @@ -62,13 +71,13 @@ def send(self, request: Request) -> Any: continue if not self.should_retry(status_code=last.status_code, attempt=attempt + 1): - return self.unwrap(last) + return self.unwrap(last, binary=request.binary) time.sleep(self.backoff(last, attempt)) # Out of attempts with a response in hand: raise what it says rather # than a generic "gave up", so the caller sees the server's own reason. assert last is not None # noqa: S101 - unreachable; the loop ran at least once - return self.unwrap(last) + return self.unwrap(last, binary=request.binary) def close(self) -> None: if self._owns_client: @@ -145,6 +154,48 @@ def list( return EmailPage.from_payload(payload) +class Inbound: + """`client.inbound` - mail your project received.""" + + def __init__(self, transport: _Transport) -> None: + self._transport = transport + + def get(self, inbound_id: str) -> InboundEmail: + """One received message, parsed. `NotFound` if it is not this key's project's.""" + return InboundEmail.from_payload(self._transport.send(build_inbound_get(inbound_id))) + + def list( + self, + *, + limit: int | None = None, + starting_after: str | None = None, + domain: str | None = None, + ) -> InboundPage: + """One page of received messages, newest first, without bodies. + + One page, not all of them. Pass `page.last_id` as `starting_after` while + `page.has_more`. + """ + payload = self._transport.send( + build_inbound_list(limit=limit, starting_after=starting_after, domain=domain) + ) + return InboundPage.from_payload(payload) + + def attachment(self, inbound_id: str, index: int) -> bytes: + """One attachment's bytes. `index` is from the message's `attachments`. + + Needs the stored original, so it raises `NotFound` once retention has + removed it. The parsed message from `get` does not expire. + """ + data: bytes = self._transport.send(build_inbound_attachment(inbound_id, index)) + return data + + def raw(self, inbound_id: str) -> bytes: + """The message exactly as it was received, as the bytes of an `.eml`.""" + data: bytes = self._transport.send(build_inbound_raw(inbound_id)) + return data + + class SesKit: """A SESKit instance, addressed over HTTP. @@ -173,6 +224,7 @@ def __init__( client=http_client, ) self.emails = Emails(self._transport) + self.inbound = Inbound(self._transport) def close(self) -> None: """Release the connection pool. Not needed for a client you kept.""" diff --git a/packages/sdk-python/src/seskit/_models.py b/packages/sdk-python/src/seskit/_models.py index aac954b..18ce855 100644 --- a/packages/sdk-python/src/seskit/_models.py +++ b/packages/sdk-python/src/seskit/_models.py @@ -135,3 +135,193 @@ def __len__(self) -> int: def last_id(self) -> str | None: """Pass as `starting_after` to fetch the next page. None when empty.""" return self.data[-1].id if self.data else None + + +# --------------------------------------------------------------- inbound --- +# +# Mail the project received. Everything in these was written by somebody else: +# nothing here is cleaned, and `html` in particular is whatever was sent. + + +@dataclass(frozen=True, slots=True, kw_only=True) +class Verdicts: + """What SES concluded about a message: `PASS`, `FAIL`, `GRAY` or + `PROCESSING_FAILED`. `None` means SES did not say, which is not a pass. + """ + + spf: str | None = None + dkim: str | None = None + dmarc: str | None = None + #: `none`, `quarantine` or `reject`, and only present when DMARC failed. + dmarc_policy: str | None = None + spam: str | None = None + virus: str | None = None + + @classmethod + def from_payload(cls, payload: dict[str, Any] | None) -> Verdicts: + data = payload or {} + return cls( + spf=data.get("spf"), + dkim=data.get("dkim"), + dmarc=data.get("dmarc"), + dmarc_policy=data.get("dmarc_policy"), + spam=data.get("spam"), + virus=data.get("virus"), + ) + + +@dataclass(frozen=True, slots=True, kw_only=True) +class InboundAttachment: + """One attachment, described. Fetch the bytes with `inbound.attachment`.""" + + index: int + filename: str + #: What the sender *declared*. It is not checked, and a download is always + #: served as `application/octet-stream`. + content_type: str + size: int + inline: bool + content_id: str | None = None + + @classmethod + def from_payload(cls, payload: dict[str, Any]) -> InboundAttachment: + return cls( + index=int(payload["index"]), + filename=str(payload.get("filename", "")), + content_type=str(payload.get("content_type", "")), + size=int(payload.get("size", 0)), + inline=bool(payload.get("inline")), + content_id=payload.get("content_id"), + ) + + +@dataclass(frozen=True, slots=True, kw_only=True) +class InboundSummary: + """A received message without its body. What `inbound.list` returns.""" + + id: str + domain: str + from_: str + from_name: str + to: list[str] + cc: list[str] + subject: str + received_at: datetime | None + size: int + attachment_count: int + verdicts: Verdicts + #: A limit was hit while reading it, so something is missing here. The + #: original has all of it. + truncated: bool + #: Nothing could be read. Everything else is empty; `inbound.raw` has the + #: original until retention removes it. + parse_failed: bool + #: Whether the original is expected to still be in storage. The parsed message + #: stays readable either way. + raw_available: bool + raw_expires_at: datetime | None + raw: dict[str, Any] = field(default_factory=dict, repr=False) + + @classmethod + def from_payload(cls, payload: dict[str, Any]) -> InboundSummary: + return cls(**_summary(payload)) + + +def _summary(payload: dict[str, Any]) -> dict[str, Any]: + """The fields a summary and a full message share.""" + return { + "id": str(payload["id"]), + "domain": str(payload.get("domain", "")), + # `from` is a keyword; the trailing underscore is the same compromise as + # `Email.from_` and `send(from_=...)`. + "from_": str(payload.get("from", "")), + "from_name": str(payload.get("from_name", "")), + "to": list(payload.get("to") or []), + "cc": list(payload.get("cc") or []), + "subject": str(payload.get("subject", "")), + "received_at": _when(payload.get("received_at")), + "size": int(payload.get("size", 0)), + "attachment_count": int(payload.get("attachment_count", 0)), + "verdicts": Verdicts.from_payload(payload.get("verdicts")), + "truncated": bool(payload.get("truncated")), + "parse_failed": bool(payload.get("parse_failed")), + "raw_available": bool(payload.get("raw_available")), + "raw_expires_at": _when(payload.get("raw_expires_at")), + "raw": payload, + } + + +@dataclass(frozen=True, slots=True, kw_only=True) +class InboundEmail(InboundSummary): + """One received message, in full. What `inbound.get` returns. + + **The bodies are the sender's.** `html` in particular may contain script, + tracking images and links that lie, and has to be treated as hostile by + anything that renders it. + """ + + text: str + html: str + envelope_from: str + #: Who it was addressed to on the wire - the addresses that matched your rule. + #: A blind copy is here and in neither `to` nor `cc`. + recipients: list[str] + reply_to: list[str] + #: Without angle brackets. Empty if the sender gave none. + message_id: str + in_reply_to: str + references: list[str] + attachments: list[InboundAttachment] + #: `(name, value)` as sent, in order, up to a limit. Raw and untrusted. + headers: list[tuple[str, str]] + + @classmethod + def from_payload(cls, payload: dict[str, Any]) -> InboundEmail: + return cls( + **_summary(payload), + text=str(payload.get("text", "")), + html=str(payload.get("html", "")), + envelope_from=str(payload.get("envelope_from", "")), + recipients=list(payload.get("recipients") or []), + reply_to=list(payload.get("reply_to") or []), + message_id=str(payload.get("message_id", "")), + in_reply_to=str(payload.get("in_reply_to", "")), + references=list(payload.get("references") or []), + attachments=[ + InboundAttachment.from_payload(a) for a in payload.get("attachments") or [] + ], + headers=[(str(h["name"]), str(h["value"])) for h in payload.get("headers") or []], + ) + + +@dataclass(frozen=True, slots=True) +class InboundPage: + """One page of `inbound.list`, newest first. + + One page and not everything, for the reason `EmailPage` is: a loop that + silently made more requests would turn one line into an unbounded number of + them against somebody's rate limit. + """ + + data: list[InboundSummary] + has_more: bool + raw: dict[str, Any] = field(default_factory=dict, repr=False) + + @classmethod + def from_payload(cls, payload: dict[str, Any]) -> InboundPage: + return cls( + data=[InboundSummary.from_payload(row) for row in payload.get("data") or []], + has_more=bool(payload.get("has_more")), + raw=payload, + ) + + def __iter__(self) -> Any: + return iter(self.data) + + def __len__(self) -> int: + return len(self.data) + + @property + def last_id(self) -> str | None: + """Pass as `starting_after` to fetch the next page. None when empty.""" + return self.data[-1].id if self.data else None diff --git a/packages/sdk-python/src/seskit/_transport.py b/packages/sdk-python/src/seskit/_transport.py index eeee121..2a79256 100644 --- a/packages/sdk-python/src/seskit/_transport.py +++ b/packages/sdk-python/src/seskit/_transport.py @@ -54,6 +54,10 @@ class Request: json: dict[str, Any] | None = None params: dict[str, Any] | None = None headers: dict[str, str] = field(default_factory=dict) + #: The answer is a file, not JSON. A download is returned as bytes exactly as + #: received - decoding it as JSON would fail, and decoding it as text would + #: corrupt it. Errors are still JSON envelopes either way. + binary: bool = False def new_idempotency_key() -> str: @@ -116,8 +120,15 @@ def headers(self, extra: dict[str, str] | None = None) -> dict[str, str]: # ------------------------------------------------------------ answers --- - def unwrap(self, response: httpx.Response) -> Any: - """The response body, or the exception its error envelope describes.""" + def unwrap(self, response: httpx.Response, *, binary: bool = False) -> Any: + """The response body, or the exception its error envelope describes. + + ``binary`` is for a download: a success is the bytes, untouched. A failure is + always a JSON envelope, so it is read the same way whichever was asked for. + """ + if binary and response.is_success: + return response.content + try: payload = response.json() except ValueError: diff --git a/packages/sdk-python/src/seskit/resources.py b/packages/sdk-python/src/seskit/resources.py index a56357b..e3f2ee6 100644 --- a/packages/sdk-python/src/seskit/resources.py +++ b/packages/sdk-python/src/seskit/resources.py @@ -16,6 +16,7 @@ import base64 from typing import Any +from urllib.parse import quote from seskit._transport import Request, new_idempotency_key @@ -24,6 +25,7 @@ Attachment = tuple[str, bytes, str] EMAILS = "/emails" +INBOUND = "/inbound" def _recipients(value: str | list[str] | None) -> list[str] | None: @@ -122,3 +124,66 @@ def build_list( if value is not None: params[name] = value return Request(method="GET", path=EMAILS, params=params or None) + + +# --------------------------------------------------------------- inbound --- + + +def _segment(value: str) -> str: + """An id as one path segment, whatever it contains. + + Ids are interpolated into a URL, and an id with a slash in it would otherwise + reach a different endpoint - ``../emails`` is a valid string. Quoting makes it + one segment that the API will say it does not know. + """ + return quote(value, safe="") + + +def build_inbound_get(inbound_id: str) -> Request: + """`GET /v1/inbound/{id}`.""" + return Request(method="GET", path=f"{INBOUND}/{_segment(inbound_id)}") + + +def build_inbound_list( + *, + limit: int | None = None, + starting_after: str | None = None, + domain: str | None = None, +) -> Request: + """`GET /v1/inbound`, newest first, without bodies. + + Cursor rather than offset, for the reason `emails.list` is: a message arriving + between two pages shifts every row of an offset down one. + """ + params: dict[str, Any] = {} + for name, value in ( + ("limit", limit), + ("starting_after", starting_after), + ("domain", domain), + ): + if value is not None: + params[name] = value + return Request(method="GET", path=INBOUND, params=params or None) + + +def build_inbound_attachment(inbound_id: str, index: int) -> Request: + """`GET /v1/inbound/{id}/attachments/{index}`, as bytes. + + ``index`` is forced through ``int`` so nothing but a number can reach the path. + """ + return Request( + method="GET", + path=f"{INBOUND}/{_segment(inbound_id)}/attachments/{int(index)}", + headers={"Accept": "*/*"}, + binary=True, + ) + + +def build_inbound_raw(inbound_id: str) -> Request: + """`GET /v1/inbound/{id}/raw`, as bytes.""" + return Request( + method="GET", + path=f"{INBOUND}/{_segment(inbound_id)}/raw", + headers={"Accept": "*/*"}, + binary=True, + ) diff --git a/tests/conftest.py b/tests/conftest.py index a0faaec..46fe3f9 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -36,12 +36,15 @@ os.environ.setdefault("SMTP_PORT", "1025") os.environ.setdefault("EMAILS_FROM_EMAIL", "tests@seskit.local") +from fakes.inbound import FakeInboundProvisioner, FakeInboundStore from fakes.queue import FakeQueue from fakes.ses import FakeProviderFactory, FakeProvisioner from httpx import ASGITransport, AsyncClient from redis.asyncio import Redis, from_url from seskit_api.dependencies import ( get_destination_resolver, + get_inbound_provisioner_factory, + get_inbound_store_factory, get_provider_factory, get_provisioner_factory, ) @@ -265,6 +268,17 @@ def provisioner_factory() -> type[FakeProvisioner]: return FakeProvisioner +@pytest.fixture +def inbound_store() -> FakeInboundStore: + """The stored-mail reader every ``app_client`` test talks to. + + Overridden for the same reason as the others: a download fetches from S3, and + no test should be able to reach AWS by forgetting to substitute something. + Seed it with ``inbound_store.put(bucket, key, raw)``. + """ + return FakeInboundStore() + + @pytest.fixture def destination_resolver() -> Callable[[str], list[str]]: """Resolves every webhook hostname to one public address. @@ -288,6 +302,7 @@ async def app_client( redis_client: Redis, provider_factory: FakeProviderFactory, provisioner_factory: type[FakeProvisioner], + inbound_store: FakeInboundStore, destination_resolver: Callable[[str], list[str]], queue: FakeQueue, ) -> AsyncIterator[AsyncClient]: @@ -308,6 +323,10 @@ async def _session() -> AsyncIterator[AsyncSession]: # asks for the `provider_factory` fixture and configures it. application.dependency_overrides[get_provider_factory] = lambda: provider_factory application.dependency_overrides[get_provisioner_factory] = lambda: provisioner_factory + application.dependency_overrides[get_inbound_provisioner_factory] = lambda: ( + FakeInboundProvisioner + ) + application.dependency_overrides[get_inbound_store_factory] = lambda: inbound_store.factory application.dependency_overrides[get_destination_resolver] = lambda: destination_resolver # The ASGI transport does not run lifespan, so app.state.queue is never # built - and a real pool would make every send need one anyway. diff --git a/tests/fakes/inbound.py b/tests/fakes/inbound.py new file mode 100644 index 0000000..7c82236 --- /dev/null +++ b/tests/fakes/inbound.py @@ -0,0 +1,84 @@ +"""Fakes for received mail, for tests that go through the real application. + +Substituted for ``S3InboundStore`` and ``SESInboundProvisioner`` so that no test +can reach AWS by forgetting to override one. They satisfy the same Protocols as +the real adapters, so a signature drift breaks here as well as in production. + +Not the fakes ``test_inbound_service`` uses. Those record calls in class-level +state to make the refcount observable; these are the quiet versions that sit +behind every ``app_client`` test, where nothing should be happening at all. +""" + +from __future__ import annotations + +from typing import ClassVar + +from seskit_core.errors import APIError, ErrorType +from seskit_core.providers import AWSCredentials, InboundInfrastructure, InboundRule + +EXPIRED = "That message is no longer in storage." + + +class FakeInboundStore: + """Stored messages by ``(bucket, key)``, and a record of every fetch.""" + + def __init__(self) -> None: + self.objects: dict[tuple[str, str], bytes] = {} + self.fetches: list[tuple[str, str]] = [] + #: Raised by every fetch when set, to stand in for a permission or a + #: network failure. + self.error: APIError | None = None + #: The region and key the last build was handed. Recorded because "does + #: this project read its mail with its own key?" is a thing a test asks. + self.built_with: tuple[str, AWSCredentials] | None = None + + def put(self, bucket: str, key: str, raw: bytes) -> None: + self.objects[(bucket, key)] = raw + + def factory(self, region: str, credentials: AWSCredentials) -> FakeInboundStore: + self.built_with = (region, credentials) + return self + + async def fetch_message(self, *, bucket: str, key: str) -> bytes: + self.fetches.append((bucket, key)) + if self.error is not None: + raise self.error + try: + return self.objects[(bucket, key)] + except KeyError: + raise APIError(ErrorType.NOT_FOUND, EXPIRED) from None + + +class FakeInboundProvisioner: + """A provisioner that builds nothing and says it did.""" + + calls: ClassVar[list[str]] = [] + + def __init__(self, region: str, credentials: AWSCredentials | None = None) -> None: + self.region = region + self.credentials = credentials + + async def provision_inbound( + self, *, bucket_name: str, topic_name: str, queue_name: str, retention_days: int + ) -> InboundInfrastructure: + FakeInboundProvisioner.calls.append("provision") + return InboundInfrastructure( + bucket=bucket_name, + topic_arn=f"arn:aws:sns:{self.region}:123456789012:{topic_name}", + queue_url=f"https://sqs.{self.region}.amazonaws.com/123456789012/{queue_name}", + queue_arn=f"arn:aws:sqs:{self.region}:123456789012:{queue_name}", + subscription_arn=f"arn:aws:sns:{self.region}:123456789012:{topic_name}:sub", + ) + + async def add_inbound_rule( + self, infrastructure: InboundInfrastructure, *, domain: str, rule_name: str + ) -> InboundRule: + FakeInboundProvisioner.calls.append("add_rule") + return InboundRule(name=rule_name, rule_set="seskit-inbound", created_rule_set=True) + + async def remove_inbound_rule(self, rule: InboundRule) -> None: + FakeInboundProvisioner.calls.append("remove_rule") + + async def remove_inbound(self, infrastructure: InboundInfrastructure) -> bool: + FakeInboundProvisioner.calls.append("remove") + return True diff --git a/tests/test_received_events_model.py b/tests/test_received_events_model.py new file mode 100644 index 0000000..400f37d --- /dev/null +++ b/tests/test_received_events_model.py @@ -0,0 +1,173 @@ +"""An event can belong to a message that arrived (inbound email, Phase C). + +Until now every event was about a message SESKit sent, so ``email_id`` was +required. Received mail has no ``emails`` row, and the webhook machinery is keyed +to an event, so an event now has exactly one of two parents. That "exactly one" is +held by a check constraint, which only a real database can test. +""" + +from __future__ import annotations + +from datetime import UTC, datetime + +import pytest +from seskit_core.events import to_public, to_public_received +from seskit_core.ids import IDPrefix, generate_id +from seskit_core.models import ( + EVENT_LABELS, + PUBLIC_EVENT_TYPES, + Email, + EmailEvent, + EventType, + Project, +) +from sqlalchemy import func, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession +from test_inbound_models import _inbound, _project + +NOW = datetime(2026, 10, 5, 12, 0, tzinfo=UTC) + + +async def _sent(session: AsyncSession, project: Project) -> Email: + email = Email( + project_id=project.id, + from_address="from@example.com", + to_addresses=["to@example.org"], + cc_addresses=[], + bcc_addresses=[], + reply_to=[], + subject="Hello", + text_body="Hi", + status="sent", + provider="ses", + ) + session.add(email) + await session.flush() + return email + + +def _event(**parents: str | None) -> EmailEvent: + event_id = generate_id(IDPrefix.EVENT) + return EmailEvent( + id=event_id, + event_type=EventType.RECEIVED.value, + occurred_at=NOW, + payload={}, + **parents, + ) + + +# ------------------------------------------------------------------- pure --- + + +def test_received_is_a_public_event_with_a_label() -> None: + assert EventType.RECEIVED in PUBLIC_EVENT_TYPES + assert EVENT_LABELS[EventType.RECEIVED] == "Received" + + +def test_the_received_payload_names_the_inbound_message_not_an_email() -> None: + """``email_id`` must never hold an inbound id: a receiver switching on the + field's presence has to be able to trust what it refers to, and + ``GET /v1/emails/{id}`` must never be handed one. + """ + payload = to_public_received( + event_id="evt_1", inbound_id="inbound_1", occurred=NOW, data={"subject": "hi"} + ) + + assert payload == { + "id": "evt_1", + "type": "email.received", + "inbound_id": "inbound_1", + "created_at": NOW.isoformat(), + "data": {"subject": "hi"}, + } + assert "email_id" not in payload + + +def test_the_sent_payload_is_unchanged_by_this() -> None: + """The contract every existing receiver relies on.""" + payload = to_public( + event_id="evt_1", + event_type=EventType.DELIVERED, + email_id="email_1", + occurred=NOW, + data={"to": ["a@example.org"]}, + ) + + assert payload["email_id"] == "email_1" + assert "inbound_id" not in payload + assert payload["type"] == "email.delivered" + + +# --------------------------------------------------------------- database --- + + +async def test_a_received_event_belongs_to_the_message_that_arrived( + db_session: AsyncSession, +) -> None: + project = await _project(db_session) + message = _inbound(project) + db_session.add(message) + await db_session.flush() + event = _event(inbound_email_id=message.id) + db_session.add(event) + await db_session.flush() + await db_session.refresh(event, ["inbound_email", "email"]) + + assert event.email_id is None + assert event.email is None + assert event.inbound_email is not None + assert event.inbound_email.id == message.id + + +async def test_a_sent_event_still_belongs_to_a_sent_email(db_session: AsyncSession) -> None: + project = await _project(db_session) + email = await _sent(db_session, project) + event = _event(email_id=email.id) + event.event_type = EventType.DELIVERED.value + db_session.add(event) + + await db_session.flush() + + assert event.inbound_email_id is None + + +async def test_an_event_about_nothing_is_refused(db_session: AsyncSession) -> None: + """Neither parent would leave an event nothing can be delivered for, since + the project is found through the parent. + """ + db_session.add(_event()) + + with pytest.raises(IntegrityError): + await db_session.flush() + + +async def test_an_event_about_two_things_is_refused(db_session: AsyncSession) -> None: + """Both would make it ambiguous which, and the webhook would name one of + them arbitrarily. + """ + project = await _project(db_session) + email = await _sent(db_session, project) + message = _inbound(project) + db_session.add(message) + await db_session.flush() + db_session.add(_event(email_id=email.id, inbound_email_id=message.id)) + + with pytest.raises(IntegrityError): + await db_session.flush() + + +async def test_deleting_a_received_message_deletes_its_events(db_session: AsyncSession) -> None: + project = await _project(db_session) + message = _inbound(project) + db_session.add(message) + await db_session.flush() + db_session.add(_event(inbound_email_id=message.id)) + await db_session.flush() + + await db_session.delete(message) + await db_session.flush() + + total = await db_session.scalar(select(func.count()).select_from(EmailEvent)) + assert total == 0 diff --git a/tests/test_received_webhooks.py b/tests/test_received_webhooks.py new file mode 100644 index 0000000..5fd987e --- /dev/null +++ b/tests/test_received_webhooks.py @@ -0,0 +1,236 @@ +"""Telling applications that mail arrived (inbound email, Phase C). + +What is asserted is the contract an integration is written against, and who gets +it. The payload is checked field by field, and then as the bytes that are signed +and sent: a payload that reads back correctly can still differ on the wire, and +the signature covers the wire. + +The isolation tests are the ones that matter most. Mail received for one project +must never produce a delivery to another project's endpoint, because the payload +carries the sender, the recipients and the subject. +""" + +from __future__ import annotations + +import json +from datetime import UTC, datetime + +from seskit_core.events import event_for, parse_received, record_received +from seskit_core.models import ( + AWSConnection, + EmailEvent, + Identity, + InboundEmail, + WebhookDelivery, + WebhookEndpoint, + WebhookStatus, +) +from seskit_core.services.webhooks import payload_bytes +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession +from test_inbound_ingest import RAW, _connection, _identity, _record, notification + +RECEIVED = datetime(2026, 10, 5, 10, 0, tzinfo=UTC) + + +async def _endpoint( + session: AsyncSession, + connection: AWSConnection, + *, + status: WebhookStatus = WebhookStatus.ACTIVE, + url: str = "https://hooks.example.com/seskit", +) -> WebhookEndpoint: + endpoint = WebhookEndpoint( + project_id=connection.project_id, + url=url, + secret="whsec_test_secret", + status=status.value, + ) + session.add(endpoint) + await session.flush() + return endpoint + + +async def _deliveries(session: AsyncSession) -> list[WebhookDelivery]: + return list(await session.scalars(select(WebhookDelivery))) + + +async def _received( + session: AsyncSession, +) -> tuple[AWSConnection, Identity, InboundEmail, EmailEvent]: + connection = await _connection(session, "owner@example.com") + identity = await _identity(session, connection) + _, row = await _record(session, identity) + assert row is not None + event = await event_for(session, row.id) + assert event is not None + return connection, identity, row, event + + +# ---------------------------------------------------------------- the event --- + + +async def test_a_received_message_raises_one_event_with_the_documented_payload( + db_session: AsyncSession, +) -> None: + _, _, row, event = await _received(db_session) + + assert event.event_type == "received" + assert event.email_id is None + assert event.inbound_email_id == row.id + assert event.provider_event_id is None + # When it happened, not when SESKit ran. + assert event.occurred_at == RECEIVED + assert event.payload == { + "id": event.id, + "type": "email.received", + "inbound_id": row.id, + "created_at": RECEIVED.isoformat(), + "data": { + "domain": "example.com", + "from": "ada@other.org", + "from_name": "Ada Lovelace", + "to": ["support@example.com"], + "cc": [], + "recipients": ["support@example.com"], + "subject": "Hello there", + "received_at": RECEIVED.isoformat(), + "size": len(RAW), + "attachment_count": 0, + "verdicts": { + "spf": "PASS", + "dkim": "GRAY", + "dmarc": "PASS", + "spam": "PASS", + "virus": "PASS", + }, + "truncated": False, + "parse_failed": False, + }, + } + assert "email_id" not in event.payload + + +async def test_the_wire_bytes_describe_the_message_and_do_not_carry_it( + db_session: AsyncSession, +) -> None: + """The signature covers these bytes, so they are what is checked. And they + must stay small: no body, no headers, no attachment, however large the mail. + """ + _, _, row, event = await _received(db_session) + + wire = payload_bytes(event) + + assert json.loads(wire) == event.payload + # Compact separators, as every other event is signed. + assert b", " not in wire + assert b": " not in wire + assert b"Hello there" in wire + assert row.text_body.strip().encode() not in wire + assert b"Message-ID" not in wire + assert b'"headers"' not in wire + + +async def test_a_verdict_ses_did_not_give_is_null_in_the_payload_never_pass( + db_session: AsyncSession, +) -> None: + """Null means "SES did not say". A receiver that filters on spam would + otherwise treat an unchecked message as a clean one. + """ + connection = await _connection(db_session, "owner@example.com") + identity = await _identity(db_session, connection) + payload = notification() + del payload["receipt"]["spfVerdict"] + del payload["receipt"]["spamVerdict"] + + _, row = await record_received( + db_session, + identity, + parse_received(payload), + RAW, + provider_event_id="sns-1", + retention_days=30, + ) + assert row is not None + event = await event_for(db_session, row.id) + assert event is not None + + verdicts = event.data["verdicts"] + assert isinstance(verdicts, dict) + assert verdicts["spf"] is None + assert verdicts["spam"] is None + assert verdicts["dkim"] == "GRAY" + + +async def test_a_duplicate_raises_no_second_event_and_no_second_delivery( + db_session: AsyncSession, +) -> None: + connection = await _connection(db_session, "owner@example.com") + identity = await _identity(db_session, connection) + await _endpoint(db_session, connection) + await _record(db_session, identity, event_id="sns-1") + await _record(db_session, identity, event_id="sns-1") + + events = await db_session.scalar(select(func.count()).select_from(EmailEvent)) + + assert events == 1 + assert len(await _deliveries(db_session)) == 1 + + +# --------------------------------------------------------------- deliveries --- + + +async def test_every_active_endpoint_of_the_project_is_queued(db_session: AsyncSession) -> None: + connection = await _connection(db_session, "owner@example.com") + identity = await _identity(db_session, connection) + first = await _endpoint(db_session, connection, url="https://a.example.com/hook") + second = await _endpoint(db_session, connection, url="https://b.example.com/hook") + _, row = await _record(db_session, identity) + assert row is not None + event = await event_for(db_session, row.id) + assert event is not None + + deliveries = await _deliveries(db_session) + + assert {d.webhook_endpoint_id for d in deliveries} == {first.id, second.id} + assert {d.event_id for d in deliveries} == {event.id} + assert all(d.status == "pending" for d in deliveries) + + +async def test_a_switched_off_endpoint_is_not_queued(db_session: AsyncSession) -> None: + connection = await _connection(db_session, "owner@example.com") + identity = await _identity(db_session, connection) + await _endpoint(db_session, connection, status=WebhookStatus.DISABLED_BY_USER) + await _endpoint(db_session, connection, status=WebhookStatus.DISABLED_AFTER_FAILURES) + + await _record(db_session, identity) + + assert await _deliveries(db_session) == [] + + +async def test_mail_for_one_project_never_reaches_another_projects_endpoint( + db_session: AsyncSession, +) -> None: + """The one that matters. The payload carries the sender, the recipients and + the subject, so a delivery to the wrong project's endpoint is a leak. + """ + mine = await _connection(db_session, "mine@example.com") + theirs = await _connection(db_session, "theirs@example.com") + identity = await _identity(db_session, mine) + my_endpoint = await _endpoint(db_session, mine) + their_endpoint = await _endpoint(db_session, theirs, url="https://theirs.example.net/hook") + + await _record(db_session, identity) + + queued = {d.webhook_endpoint_id for d in await _deliveries(db_session)} + assert queued == {my_endpoint.id} + assert their_endpoint.id not in queued + + +async def test_a_project_with_no_endpoints_raises_the_event_and_queues_nothing( + db_session: AsyncSession, +) -> None: + _, _, _, event = await _received(db_session) + + assert event is not None + assert await _deliveries(db_session) == [] diff --git a/tests/test_sdk_inbound_contract.py b/tests/test_sdk_inbound_contract.py new file mode 100644 index 0000000..97a0277 --- /dev/null +++ b/tests/test_sdk_inbound_contract.py @@ -0,0 +1,192 @@ +"""The SDK's inbound calls against the real API (inbound email, Phase C). + +``tests/test_sdk_inbound_transport.py`` asserts the client sends what it thinks it +sends. This asserts that what it thinks is right: the same client, pointed at the +actual application, with the actual Postgres behind it. A stub written from the +author's memory of the API agrees with a misremembering for ever; here a wrong +field name is a missing attribute and a renamed error type is an exception nobody +can catch. + +Like ``test_sdk_contract`` it uses the async client, because ``httpx.ASGITransport`` +is async-only. The two clients share the builders, the URL, the headers and the +transport rule, so what is proved here holds for both. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +import pytest +from fakes.inbound import FakeInboundStore +from httpx import AsyncClient +from seskit import InboundEmail, InboundSummary, NotFound +from sqlalchemy.ext.asyncio import AsyncSession +from test_sdk_contract import _key, _sdk +from test_v1_inbound import _message +from test_v1_inbound_downloads import PDF, _raw, _stored + + +async def test_the_list_reads_back_what_the_api_sent( + app_client: AsyncClient, db_session: AsyncSession +) -> None: + project_id, key = await _key(db_session) + await _message(db_session, project_id=project_id, subject="Hello there") + await db_session.commit() + + page = await _sdk(app_client, key).inbound.list() + + assert len(page) == 1 + (row,) = page.data + assert isinstance(row, InboundSummary) + assert not isinstance(row, InboundEmail) + assert row.subject == "Hello there" + assert row.from_ == "ada@other.org" + assert row.to == ["support@example.com"] + assert row.verdicts.spf == "PASS" + assert row.verdicts.dkim == "GRAY" + # Left null by the API, so None here - not a pass. + assert row.verdicts.virus is None + assert row.received_at is not None + assert row.received_at.tzinfo is not None + assert page.has_more is False + + +async def test_one_message_reads_back_in_full( + app_client: AsyncClient, db_session: AsyncSession +) -> None: + project_id, key = await _key(db_session) + row = await _message( + db_session, + project_id=project_id, + envelope_to=["support@example.com", "bcc@example.com"], + message_id_header="abc@other.org", + reference_ids=["root@example.com"], + attachments=[ + { + "index": 0, + "filename": "report.pdf", + "content_type": "application/pdf", + "size": 9, + "inline": False, + "content_id": None, + } + ], + headers=[["Subject", "Hello there"], ["X-Odd", "1"]], + ) + await db_session.commit() + + message = await _sdk(app_client, key).inbound.get(row.id) + + assert isinstance(message, InboundEmail) + assert message.text == "Hi." + assert message.html == "
Hi.
" + assert message.recipients == ["support@example.com", "bcc@example.com"] + assert message.message_id == "abc@other.org" + assert message.references == ["root@example.com"] + assert [a.filename for a in message.attachments] == ["report.pdf"] + assert message.attachments[0].index == 0 + assert message.headers == [("Subject", "Hello there"), ("X-Odd", "1")] + assert message.raw_available is True + + +async def test_paging_by_last_id_neither_skips_nor_repeats( + app_client: AsyncClient, db_session: AsyncSession +) -> None: + project_id, key = await _key(db_session) + for n in range(5): + await _message(db_session, project_id=project_id, second=n, subject=f"m{n}") + await db_session.commit() + inbound = _sdk(app_client, key).inbound + + seen: list[str] = [] + page = await inbound.list(limit=2) + seen += [row.subject for row in page] + while page.has_more: + page = await inbound.list(limit=2, starting_after=page.last_id) + seen += [row.subject for row in page] + + assert seen == ["m4", "m3", "m2", "m1", "m0"] + + +async def test_the_domain_filter_reaches_the_api( + app_client: AsyncClient, db_session: AsyncSession +) -> None: + project_id, key = await _key(db_session) + await _message(db_session, project_id=project_id, domain="example.com", second=1) + await _message(db_session, project_id=project_id, domain="example.net", second=2) + await db_session.commit() + + page = await _sdk(app_client, key).inbound.list(domain="example.net") + + assert [row.domain for row in page] == ["example.net"] + + +async def test_another_projects_message_is_not_found( + app_client: AsyncClient, db_session: AsyncSession +) -> None: + _, mine = await _key(db_session, owner="mine@example.com") + theirs_project, _ = await _key(db_session, owner="theirs@example.com") + foreign = await _message(db_session, project_id=theirs_project) + await db_session.commit() + + with pytest.raises(NotFound): + await _sdk(app_client, mine).inbound.get(foreign.id) + + +async def test_an_id_that_tries_to_reach_another_endpoint_is_just_not_found( + app_client: AsyncClient, db_session: AsyncSession +) -> None: + """Against the real API, not a stub: ``../emails`` has to arrive as one + segment that nothing matches, and not as a different endpoint's answer. + """ + _, key = await _key(db_session) + await db_session.commit() + + with pytest.raises(NotFound): + await _sdk(app_client, key).inbound.get("../emails") + + +# ------------------------------------------------------------- downloads --- + + +async def test_an_attachment_and_the_original_come_back_byte_for_byte( + app_client: AsyncClient, db_session: AsyncSession, inbound_store: FakeInboundStore +) -> None: + """The bytes survive a trip through the real API and the real transport: no + decoding as text, no JSON, nothing between a PDF in a bucket and a PDF in the + caller's hands. + """ + _, key, row = await _stored(db_session, inbound_store, owner="owner@example.com") + inbound = _sdk(app_client, key).inbound + + assert await inbound.attachment(row.id, 0) == PDF + assert await inbound.raw(row.id) == _raw() + + +async def test_an_attachment_that_does_not_exist_is_not_found( + app_client: AsyncClient, db_session: AsyncSession, inbound_store: FakeInboundStore +) -> None: + _, key, row = await _stored(db_session, inbound_store, owner="owner@example.com") + + with pytest.raises(NotFound): + await _sdk(app_client, key).inbound.attachment(row.id, 9) + + +async def test_a_download_after_retention_is_not_found_but_the_message_still_reads( + app_client: AsyncClient, db_session: AsyncSession, inbound_store: FakeInboundStore +) -> None: + _, key, row = await _stored( + db_session, + inbound_store, + owner="owner@example.com", + raw_expires_at=datetime.now(UTC) - timedelta(days=1), + ) + inbound = _sdk(app_client, key).inbound + + with pytest.raises(NotFound) as caught: + await inbound.raw(row.id) + + assert "no longer in storage" in str(caught.value) + message = await inbound.get(row.id) + assert message.text == "Hi." + assert message.raw_available is False diff --git a/tests/test_sdk_inbound_transport.py b/tests/test_sdk_inbound_transport.py new file mode 100644 index 0000000..a616935 --- /dev/null +++ b/tests/test_sdk_inbound_transport.py @@ -0,0 +1,300 @@ +"""The SDK's inbound calls, against a stub server (inbound email, Phase C). + +What a request looks like and what a response becomes, with no real server, so +these run anywhere and can assert on the exact bytes that would have gone out. +``tests/test_sdk_inbound_contract.py`` is the other half: the same client pointed +at the real application. + +Two things are specific to this resource. A download is *bytes*, and the transport +has to hand them back untouched - decoding them as JSON fails and decoding them as +text corrupts them. And an id is interpolated into a URL, so what an id can reach +is worth a test of its own. +""" + +from __future__ import annotations + +from typing import Any + +import httpx +import pytest +from seskit import ( + AsyncSesKit, + InboundEmail, + InboundPage, + InboundSummary, + NotFound, + SesKit, + Verdicts, +) +from test_sdk_transport import API_KEY, BASE_URL, Recorder, _client, _error, _ok + +SUMMARY: dict[str, Any] = { + "id": "inbound_01J8XQ", + "domain": "example.com", + "from": "ada@other.org", + "from_name": "Ada Lovelace", + "to": ["support@example.com"], + "cc": [], + "subject": "Hello there", + "received_at": "2026-10-05T10:00:00+00:00", + "size": 4821, + "attachment_count": 1, + "verdicts": {"spf": "PASS", "dkim": "GRAY", "dmarc": None, "spam": "PASS", "virus": None}, + "truncated": False, + "parse_failed": False, + "raw_available": True, + "raw_expires_at": "2026-11-04T10:00:00+00:00", +} + +FULL: dict[str, Any] = { + **SUMMARY, + "text": "Hi.", + "html": "Hi.
", + "envelope_from": "bounce@other.org", + "recipients": ["support@example.com", "bcc@example.com"], + "reply_to": ["replies@other.org"], + "message_id": "abc@other.org", + "in_reply_to": "first@example.com", + "references": ["root@example.com", "first@example.com"], + "attachments": [ + { + "index": 0, + "filename": "report.pdf", + "content_type": "application/pdf", + "size": 9, + "inline": False, + "content_id": None, + } + ], + "headers": [{"name": "Subject", "value": "Hello there"}], +} + +#: Not text, not UTF-8, with a NUL in it: what a PDF looks like to a decoder. +BINARY = b"%PDF-1.4\n\x00\xff\xfe\x80binary\r\n" + + +def _bytes(content: bytes = BINARY, status_code: int = 200) -> httpx.Response: + return httpx.Response(status_code, content=content) + + +# --------------------------------------------------------------- reading --- + + +def test_a_received_message_becomes_an_inbound_email() -> None: + recorder = Recorder(_ok(FULL)) + + message = _client(recorder).inbound.get("inbound_01J8XQ") + + assert isinstance(message, InboundEmail) + assert recorder.requests[0].method == "GET" + assert recorder.requests[0].url.path == "/v1/inbound/inbound_01J8XQ" + assert message.id == "inbound_01J8XQ" + # `from` is a keyword, so the attribute carries a trailing underscore. + assert message.from_ == "ada@other.org" + assert message.text == "Hi." + assert message.html == "Hi.
" + assert message.recipients == ["support@example.com", "bcc@example.com"] + assert message.references == ["root@example.com", "first@example.com"] + assert message.attachments[0].filename == "report.pdf" + assert message.attachments[0].index == 0 + assert message.headers == [("Subject", "Hello there")] + assert message.received_at is not None + assert message.received_at.year == 2026 + + +def test_a_verdict_the_api_left_null_stays_none_and_is_never_a_pass() -> None: + message = _client(Recorder(_ok(FULL))).inbound.get("inbound_01J8XQ") + + assert message.verdicts == Verdicts( + spf="PASS", dkim="GRAY", dmarc=None, spam="PASS", virus=None + ) + assert message.verdicts.virus is None + assert message.verdicts.dmarc_policy is None + + +def test_a_page_is_summaries_and_carries_no_body() -> None: + recorder = Recorder( + _ok({"data": [SUMMARY, {**SUMMARY, "id": "inbound_01J8XR"}], "has_more": True}) + ) + + page = _client(recorder).inbound.list() + + assert isinstance(page, InboundPage) + assert len(page) == 2 + assert page.has_more is True + assert page.last_id == "inbound_01J8XR" + assert all(isinstance(row, InboundSummary) for row in page) + assert not any(isinstance(row, InboundEmail) for row in page) + assert not hasattr(page.data[0], "html") + + +def test_an_empty_page_has_no_last_id() -> None: + page = _client(Recorder(_ok({"data": [], "has_more": False}))).inbound.list() + + assert page.last_id is None + assert len(page) == 0 + + +def test_list_sends_only_the_filters_it_was_given() -> None: + recorder = Recorder(_ok({"data": [], "has_more": False})) + client = _client(recorder) + + client.inbound.list() + client.inbound.list(limit=5, starting_after="inbound_x", domain="example.com") + + assert dict(recorder.requests[0].url.params) == {} + assert dict(recorder.requests[1].url.params) == { + "limit": "5", + "starting_after": "inbound_x", + "domain": "example.com", + } + + +def test_a_field_this_version_does_not_know_is_kept() -> None: + """A client from before a field existed still hands it to the caller.""" + message = _client(Recorder(_ok({**FULL, "thread_id": "t_1"}))).inbound.get("inbound_01J8XQ") + + assert message.raw["thread_id"] == "t_1" + + +def test_a_sparse_payload_still_reads() -> None: + message = _client(Recorder(_ok({"id": "inbound_1"}))).inbound.get("inbound_1") + + assert message.id == "inbound_1" + assert message.attachments == [] + assert message.headers == [] + assert message.verdicts == Verdicts() + assert message.received_at is None + + +# ------------------------------------------------------------- downloads --- + + +def test_an_attachment_is_the_bytes_untouched() -> None: + """Not decoded as JSON, which would fail, and not as text, which would + corrupt it: a NUL, an 0xff and a lone 0x80 all have to survive. + """ + recorder = Recorder(_bytes()) + + data = _client(recorder).inbound.attachment("inbound_01J8XQ", 0) + + assert data == BINARY + assert isinstance(data, bytes) + request = recorder.requests[0] + assert request.url.path == "/v1/inbound/inbound_01J8XQ/attachments/0" + # A download does not claim to want JSON. + assert request.headers["Accept"] == "*/*" + assert request.headers["Authorization"] == f"Bearer {API_KEY}" + + +def test_the_original_is_the_bytes_untouched() -> None: + recorder = Recorder(_bytes(b"From: a@b.org\r\n\r\n\xff\x00body")) + + data = _client(recorder).inbound.raw("inbound_01J8XQ") + + assert data == b"From: a@b.org\r\n\r\n\xff\x00body" + assert recorder.requests[0].url.path == "/v1/inbound/inbound_01J8XQ/raw" + + +def test_a_download_that_fails_is_still_a_typed_error() -> None: + """Errors are JSON envelopes whether or not a file was asked for.""" + recorder = Recorder(_error(404, "not_found", "That message is no longer in storage.")) + + with pytest.raises(NotFound) as caught: + _client(recorder).inbound.raw("inbound_01J8XQ") + + assert "no longer in storage" in str(caught.value) + + +def test_a_download_is_tried_again_on_a_server_error() -> None: + """It shares the retry rule with every other call.""" + recorder = Recorder(_error(500, "internal_error"), _bytes()) + + assert _client(recorder).inbound.raw("inbound_01J8XQ") == BINARY + assert len(recorder.requests) == 2 + + +def test_a_refusal_is_not_retried_when_a_file_was_asked_for() -> None: + recorder = Recorder(_error(404, "not_found"), _bytes()) + + with pytest.raises(NotFound): + _client(recorder).inbound.attachment("inbound_01J8XQ", 0) + + assert len(recorder.requests) == 1 + + +# ------------------------------------------------- what an id can reach --- + + +def test_an_id_cannot_leave_its_path_segment() -> None: + """Ids go into a URL, and ``../emails`` is a valid string. Quoted, it stays + one segment, which the API will say it does not know - and never a different + endpoint answering with a message that parses as the wrong thing. + """ + recorder = Recorder(_error(404, "not_found")) + + with pytest.raises(NotFound): + _client(recorder).inbound.get("../emails") + + assert recorder.requests[0].url.raw_path == b"/v1/inbound/..%2Femails" + + +@pytest.mark.parametrize("bad_id", ["a/b", "a?x=1", "a#frag", "a b", "\r\nX: y"]) +def test_no_id_adds_a_segment_a_query_or_a_header(bad_id: str) -> None: + recorder = Recorder(_error(404, "not_found")) + + with pytest.raises(NotFound): + _client(recorder).inbound.raw(bad_id) + + request = recorder.requests[0] + assert request.url.raw_path.count(b"/") == 4 + assert request.url.query == b"" + assert request.url.fragment == "" + assert "X" not in request.headers + + +def test_an_index_that_is_not_a_number_never_reaches_the_url() -> None: + recorder = Recorder(_bytes()) + + with pytest.raises(ValueError): + _client(recorder).inbound.attachment("inbound_01J8XQ", "0/../../raw") # type: ignore[arg-type] + + assert recorder.requests == [] + + +# ------------------------------------------------- the two clients agree --- + + +async def test_the_async_client_sends_the_same_request_and_returns_the_same_bytes() -> None: + """What is proved here and in the contract file holds for both: the request + builders, the URL, the headers and the transport rule are shared. + """ + sync_recorder = Recorder(_bytes()) + async_recorder = Recorder(_bytes()) + sync_client: SesKit = _client(sync_recorder) + async_client = AsyncSesKit( + api_key=API_KEY, + base_url=BASE_URL, + http_client=httpx.AsyncClient(transport=httpx.MockTransport(async_recorder)), + ) + + from_sync = sync_client.inbound.attachment("inbound_01J8XQ", 2) + from_async = await async_client.inbound.attachment("inbound_01J8XQ", 2) + + assert from_sync == from_async == BINARY + one, two = sync_recorder.requests[0], async_recorder.requests[0] + assert (one.method, one.url, one.content) == (two.method, two.url, two.content) + assert one.headers["Accept"] == two.headers["Accept"] == "*/*" + + +async def test_the_async_client_reads_a_message_the_same_way() -> None: + client = AsyncSesKit( + api_key=API_KEY, + base_url=BASE_URL, + http_client=httpx.AsyncClient(transport=httpx.MockTransport(Recorder(_ok(FULL)))), + ) + + message = await client.inbound.get("inbound_01J8XQ") + + assert message.subject == "Hello there" + assert message.attachments[0].filename == "report.pdf" diff --git a/tests/test_security_headers.py b/tests/test_security_headers.py index ee6da6e..9a191a8 100644 --- a/tests/test_security_headers.py +++ b/tests/test_security_headers.py @@ -19,8 +19,10 @@ import re from pathlib import Path +from typing import Any import pytest +from fastapi import Response from httpx import AsyncClient from seskit_api.middleware.security import BASE_POLICY, HSTS, policy_for @@ -217,3 +219,32 @@ def test_the_base_policy_leaves_scripts_to_be_completed() -> None: all, because it would look complete. """ assert "script-src" not in BASE_POLICY + + +async def test_a_route_that_sets_its_own_policy_keeps_it(app: Any, client: AsyncClient) -> None: + """Found by a test for something else. A download of somebody's file sets a + `sandbox` policy, and the middleware overwrote it with the dashboard's, which + allows `script-src 'self'`: the stricter header was discarded and the looser + one shipped, with nothing anywhere to say so. + """ + + async def strict() -> Response: + return Response("x", headers={"Content-Security-Policy": "sandbox; default-src 'none'"}) + + app.add_api_route("/_strict_policy", strict) + + response = await client.get("/_strict_policy") + + assert response.headers["Content-Security-Policy"] == "sandbox; default-src 'none'" + # Only the policy yields; the rest of the headers are still applied. + assert response.headers["X-Frame-Options"] == "DENY" + assert response.headers["X-Content-Type-Options"] == "nosniff" + + +async def test_a_route_that_sets_no_policy_still_gets_the_dashboards(client: AsyncClient) -> None: + """The other half: letting a route override must not leave every other one + without a policy. + """ + policy = (await client.get("/healthz")).headers["Content-Security-Policy"] + + assert "script-src 'self' 'nonce-" in policy diff --git a/tests/test_v1_inbound.py b/tests/test_v1_inbound.py new file mode 100644 index 0000000..b350082 --- /dev/null +++ b/tests/test_v1_inbound.py @@ -0,0 +1,458 @@ +"""``GET /v1/inbound`` and ``GET /v1/inbound/{id}`` (inbound email, Phase C). + +The read side of receiving. As with ``/v1/emails``, paging is where a list +endpoint goes wrong quietly - a wrong page returns real rows in a plausible order +and leaves some out - so a good part of this file is about the boundary: that a +cursor from another project cannot position a page, and that an id belonging to +somebody else is a 404 and never a 403, which would confirm it exists. + +The other thing worth asserting is what is *not* in a response. The storage +bucket and key are internal and never leave; the summary carries no body. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from typing import Any + +import pytest +from httpx import AsyncClient +from seskit_core.models import InboundEmail +from seskit_core.services import create_api_key, create_project, register_user +from sqlalchemy.ext.asyncio import AsyncSession +from ulid import ULID + +PASSWORD = "correct-horse-battery" +URL = "/v1/inbound" +FIRST_SECOND = 1_700_000_000 +NOW = datetime(2026, 10, 5, 12, 0, tzinfo=UTC) + + +def _auth(raw_key: str) -> dict[str, str]: + return {"Authorization": f"Bearer {raw_key}"} + + +async def _project_key( + session: AsyncSession, *, owner: str, name: str = "Inbox" +) -> tuple[str, str]: + user = await register_user(session, email=owner, password=PASSWORD, allow_signup=True) + project = await create_project(session, user_id=user.id, name=name) + issued = await create_api_key(session, project_id=project.id, name="prod") + return project.id, issued.raw_key + + +async def _message( + session: AsyncSession, + *, + project_id: str, + second: int = 0, + domain: str = "example.com", + subject: str = "Hello there", + **overrides: Any, +) -> InboundEmail: + fields: dict[str, Any] = { + "id": f"inbound_{ULID.from_timestamp(FIRST_SECOND + second)}", + "project_id": project_id, + "domain": domain, + "provider_message_id": f"ses-{second}", + "provider_event_id": f"sns-{second}-{project_id}", + "storage_bucket": "seskit-inbound-123456789012-us-east-1", + "storage_key": f"seskit-abc/{second}-{project_id}", + "received_at": NOW, + "raw_expires_at": NOW + timedelta(days=36500), + "from_address": "ada@other.org", + "from_name": "Ada Lovelace", + "to_addresses": ["support@example.com"], + "subject": subject, + "text_body": "Hi.", + "html_body": "Hi.
", + "spf_verdict": "PASS", + "dkim_verdict": "GRAY", + } + row = InboundEmail(**(fields | overrides)) + session.add(row) + await session.flush() + return row + + +# ------------------------------------------------------------------- auth --- + + +@pytest.mark.parametrize("path", [URL, f"{URL}/inbound_x"]) +async def test_a_request_without_a_key_is_refused(app_client: AsyncClient, path: str) -> None: + response = await app_client.get(path) + + assert response.status_code == 401 + assert response.json()["error"]["type"] == "authentication_failed" + + +# ------------------------------------------------------------------- list --- + + +async def test_an_empty_project_lists_nothing_rather_than_failing( + app_client: AsyncClient, db_session: AsyncSession +) -> None: + _, key = await _project_key(db_session, owner="a@example.com") + await db_session.commit() + + response = await app_client.get(URL, headers=_auth(key)) + + assert response.status_code == 200 + assert response.json() == {"data": [], "has_more": False} + + +async def test_the_list_describes_each_message_and_carries_no_body( + app_client: AsyncClient, db_session: AsyncSession +) -> None: + project_id, key = await _project_key(db_session, owner="a@example.com") + row = await _message(db_session, project_id=project_id) + await db_session.commit() + + (item,) = (await app_client.get(URL, headers=_auth(key))).json()["data"] + + assert item["id"] == row.id + assert item["from"] == "ada@other.org" + assert item["to"] == ["support@example.com"] + assert item["subject"] == "Hello there" + assert item["domain"] == "example.com" + assert item["verdicts"]["spf"] == "PASS" + assert item["verdicts"]["dkim"] == "GRAY" + # SES did not say, which is null and not a pass. + assert item["verdicts"]["virus"] is None + assert item["attachment_count"] == 0 + # A page of a hundred messages with their HTML in it is not what a list is for. + for absent in ("text", "html", "headers", "attachments"): + assert absent not in item + + +async def test_nothing_internal_ever_appears_in_a_response( + app_client: AsyncClient, db_session: AsyncSession +) -> None: + """The bucket and the key say where a customer's mail is stored. Neither is + the caller's to know, and neither is any use to them. + """ + project_id, key = await _project_key(db_session, owner="a@example.com") + row = await _message(db_session, project_id=project_id) + await db_session.commit() + + listed = (await app_client.get(URL, headers=_auth(key))).text + detail = (await app_client.get(f"{URL}/{row.id}", headers=_auth(key))).text + + for body in (listed, detail): + assert "seskit-inbound-123456789012" not in body + assert "storage_key" not in body + assert "storage_bucket" not in body + assert row.storage_key not in body + assert "project_id" not in body + + +async def test_another_projects_mail_is_never_listed( + app_client: AsyncClient, db_session: AsyncSession +) -> None: + mine, key = await _project_key(db_session, owner="a@example.com") + theirs, _ = await _project_key(db_session, owner="b@example.com") + await _message(db_session, project_id=mine, subject="mine") + await _message(db_session, project_id=theirs, subject="theirs", second=1) + await db_session.commit() + + subjects = [ + m["subject"] for m in (await app_client.get(URL, headers=_auth(key))).json()["data"] + ] + + assert subjects == ["mine"] + + +async def test_the_list_is_newest_first_and_pages_by_cursor( + app_client: AsyncClient, db_session: AsyncSession +) -> None: + project_id, key = await _project_key(db_session, owner="a@example.com") + ids = [(await _message(db_session, project_id=project_id, second=n)).id for n in range(5)] + await db_session.commit() + newest_first = list(reversed(ids)) + + first = (await app_client.get(URL, params={"limit": 2}, headers=_auth(key))).json() + second = ( + await app_client.get( + URL, + params={"limit": 2, "starting_after": first["data"][-1]["id"]}, + headers=_auth(key), + ) + ).json() + third = ( + await app_client.get( + URL, + params={"limit": 2, "starting_after": second["data"][-1]["id"]}, + headers=_auth(key), + ) + ).json() + + assert [m["id"] for m in first["data"]] == newest_first[:2] + assert first["has_more"] is True + assert [m["id"] for m in second["data"]] == newest_first[2:4] + assert second["has_more"] is True + assert [m["id"] for m in third["data"]] == newest_first[4:] + # The last page says so, and costs no request that comes back empty. + assert third["has_more"] is False + + +async def test_a_page_is_stable_while_mail_arrives_underneath_it( + app_client: AsyncClient, db_session: AsyncSession +) -> None: + """The reason for a cursor over an offset. With an offset, a message arriving + between two pages shifts every row down one and the reader silently skips + the one that moved across the boundary. + """ + project_id, key = await _project_key(db_session, owner="a@example.com") + for n in range(4): + await _message(db_session, project_id=project_id, second=n) + await db_session.commit() + first = (await app_client.get(URL, params={"limit": 2}, headers=_auth(key))).json() + + await _message(db_session, project_id=project_id, second=100) + await db_session.commit() + + second = ( + await app_client.get( + URL, + params={"limit": 2, "starting_after": first["data"][-1]["id"]}, + headers=_auth(key), + ) + ).json() + seen = [m["id"] for m in first["data"]] + [m["id"] for m in second["data"]] + assert len(seen) == len(set(seen)) == 4 + + +async def test_a_cursor_from_another_project_is_a_404_not_a_page( + app_client: AsyncClient, db_session: AsyncSession +) -> None: + """The comparison is lexical, so a foreign id would otherwise position a page + of real messages by a value the caller cannot see - a wrong answer that + looks like a right one. + """ + mine, key = await _project_key(db_session, owner="a@example.com") + theirs, _ = await _project_key(db_session, owner="b@example.com") + await _message(db_session, project_id=mine) + foreign = await _message(db_session, project_id=theirs, second=50) + await db_session.commit() + + response = await app_client.get(URL, params={"starting_after": foreign.id}, headers=_auth(key)) + + assert response.status_code == 404 + assert response.json()["error"]["type"] == "not_found" + + +async def test_an_unknown_cursor_is_a_404_not_an_empty_page( + app_client: AsyncClient, db_session: AsyncSession +) -> None: + _, key = await _project_key(db_session, owner="a@example.com") + await db_session.commit() + + response = await app_client.get( + URL, params={"starting_after": "inbound_nope"}, headers=_auth(key) + ) + + assert response.status_code == 404 + + +async def test_the_domain_filter_narrows_the_list_and_ignores_case( + app_client: AsyncClient, db_session: AsyncSession +) -> None: + project_id, key = await _project_key(db_session, owner="a@example.com") + await _message(db_session, project_id=project_id, domain="example.com", second=1) + await _message(db_session, project_id=project_id, domain="example.net", second=2) + await db_session.commit() + + response = await app_client.get(URL, params={"domain": " Example.NET "}, headers=_auth(key)) + + assert [m["domain"] for m in response.json()["data"]] == ["example.net"] + + +@pytest.mark.parametrize("limit", [0, 101, -1]) +async def test_a_page_size_out_of_range_is_refused( + app_client: AsyncClient, db_session: AsyncSession, limit: int +) -> None: + _, key = await _project_key(db_session, owner="a@example.com") + await db_session.commit() + + response = await app_client.get(URL, params={"limit": limit}, headers=_auth(key)) + + assert response.status_code in (400, 422) + assert "error" in response.json() + + +# ----------------------------------------------------------------- detail --- + + +async def test_one_message_comes_back_in_full( + app_client: AsyncClient, db_session: AsyncSession +) -> None: + project_id, key = await _project_key(db_session, owner="a@example.com") + row = await _message( + db_session, + project_id=project_id, + envelope_from="bounce@other.org", + envelope_to=["support@example.com", "bcc@example.com"], + reply_to=["replies@other.org"], + message_id_header="abc@other.org", + in_reply_to="first@example.com", + reference_ids=["root@example.com", "first@example.com"], + attachments=[ + { + "index": 0, + "filename": "report.pdf", + "content_type": "application/pdf", + "size": 9, + "inline": False, + "content_id": None, + } + ], + headers=[["Subject", "Hello there"], ["X-Odd", "1"]], + size_bytes=1234, + ) + await db_session.commit() + + body = (await app_client.get(f"{URL}/{row.id}", headers=_auth(key))).json() + + assert body["id"] == row.id + assert body["text"] == "Hi." + assert body["html"] == "Hi.
" + assert body["envelope_from"] == "bounce@other.org" + # A blind copy is in the envelope list and in neither header list. + assert body["recipients"] == ["support@example.com", "bcc@example.com"] + assert body["to"] == ["support@example.com"] + assert body["message_id"] == "abc@other.org" + assert body["in_reply_to"] == "first@example.com" + assert body["references"] == ["root@example.com", "first@example.com"] + assert body["reply_to"] == ["replies@other.org"] + assert body["attachments"] == [ + { + "index": 0, + "filename": "report.pdf", + "content_type": "application/pdf", + "size": 9, + "inline": False, + "content_id": None, + } + ] + assert body["attachment_count"] == 1 + assert body["headers"] == [ + {"name": "Subject", "value": "Hello there"}, + {"name": "X-Odd", "value": "1"}, + ] + assert body["size"] == 1234 + + +async def test_hostile_content_is_returned_exactly_as_sent( + app_client: AsyncClient, db_session: AsyncSession +) -> None: + """The API describes what arrived; it does not clean it. A consumer that + renders it has to treat it as hostile, and the field says so. + """ + project_id, key = await _project_key(db_session, owner="a@example.com") + hostile = '