diff --git a/.github/scripts/check_strict_types.sh b/.github/scripts/check_strict_types.sh index 4855a9ec9..d1647516b 100755 --- a/.github/scripts/check_strict_types.sh +++ b/.github/scripts/check_strict_types.sh @@ -46,10 +46,11 @@ _ = (flag_value, all_flags, enabled, payload, evaluations, active) emails = Emails("phs_example") template_send: SendEmailResult = emails.send(to="user@example.com", template_id="0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b", template_data={"name": "Ada"}) rendered_send: SendEmailResult = emails.send(to="user@example.com", template_key="password-reset", subject="Reset", html="

Reset

", from_="Acme ", reply_to=["help@example.com"], capture_engagement=False, max_retries=0) +invite_send: SendEmailResult = emails.send(to="user@example.com", template_key="invite", subject="Invite", html="

Invite

", cc="team@example.com", bcc=["audit@example.com"], attachments=[{"filename": "invite.ics", "content": b"BEGIN:VCALENDAR", "content_type": "text/calendar"}, {"filename": "logo.png", "content": "iVBORw0KGgo=", "content_id": "logo"}]) tracked_send: SendEmailResult = emails.send(to="user@example.com", template_key="welcome", subject="Welcome", html="

Hi

", tracking={"clicks": True, "engagement": False}) untracked_send: SendEmailResult = emails.send(to="user@example.com", template_key="welcome", subject="Welcome", html="

Hi

", tracking=False) not_sensitive_send: SendEmailResult = emails.send(to="user@example.com", template_key="newsletter", subject="News", html="

News

", sensitive=False) -_ = (template_send, rendered_send, tracked_send, untracked_send) +_ = (template_send, rendered_send, invite_send, tracked_send, untracked_send, not_sensitive_send) async def async_email_types() -> None: @@ -113,7 +114,9 @@ Emails("phs_example").send(to="user@example.com", template_id="0190f6d2-7c1e-7b3 Emails("phs_example").send(to="user@example.com", template_id="0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b", sensitive=False) # expected-type-error Emails("phs_example").send(to="user@example.com", template_key="welcome", subject="Hi", html="

Hi

", tracking={"open": False}) # expected-type-error Emails("phs_example").send(to="user@example.com", template_key="welcome", subject="Hi", html="

Hi

", template_data={"name": "Ada"}) # expected-type-error -Emails("phs_example").send(to="user@example.com", template_key="welcome", subject="Hi", html="

Hi

", cc="team@example.com") # expected-type-error +Emails("phs_example").send(to="user@example.com", template_key="welcome", subject="Hi", html="

Hi

", attachments=[{"filename": "a.txt", "content": [104, 105]}]) # expected-type-error +Emails("phs_example").send(to="user@example.com", template_key="welcome", subject="Hi", html="

Hi

", attachments=[{"filename": "a.txt", "content": b"hi", "path": "/tmp/a.txt"}]) # expected-type-error +Emails("phs_example").send(to="user@example.com", template_id="0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b", attachments=[{"filename": "a.txt", "content": b"hi"}]) # expected-type-error async def invalid_async_email() -> None: diff --git a/.sampo/changesets/emails-code-attachments.md b/.sampo/changesets/emails-code-attachments.md new file mode 100644 index 000000000..48bc79ed0 --- /dev/null +++ b/.sampo/changesets/emails-code-attachments.md @@ -0,0 +1,5 @@ +--- +pypi/posthog: minor +--- + +A code send with `posthog.emails.send` takes `cc`, `bcc` and `attachments`: at most 10 files and 40 MB per message after Base64 encoding, about 29 MB of files, with the HTML counted twice when there is no `text`, since PostHog derives the text from it. An attachment is a `filename` and its `content`, as bytes or a Base64 string, with an optional `content_type` and `content_id`. The client checks the limits before any request, uploads each file PostHog doesn't hold yet straight to PostHog's file storage over https, never following a redirect, then sends the email with each file's SHA-256. A refused or redirected upload raises `EmailError` with the code `attachment_upload_failed`, and an unreachable storage `network_error` with `field` naming the attachment. Either way no email is sent. diff --git a/posthog/emails.py b/posthog/emails.py index 2839a2dd6..141255b17 100644 --- a/posthog/emails.py +++ b/posthog/emails.py @@ -59,6 +59,19 @@ ``"resend"``, on the client or on :class:`Emails`, so PostHog knows where your sends came from. +Attachments: a code send takes ``attachments``, at most 10 files and 40 MB per +message after Base64 encoding, about 29 MB of files. Without ``text``, the +HTML counts twice, since PostHog derives the text from it. The client checks +both before any request. It then asks PostHog which files it lacks, uploads +each of those to the https URL PostHog answers, without the secret key, and +sends the email naming each file by its SHA-256. A file the project already +sent uploads nothing, and the same bytes in two attachments upload once. An +upload retries like a send when the storage is unreachable or answers a +passing fault. A failed upload sends no email. A plain HTTP upload URL is +refused before any upload unless the client's host is plain HTTP too, as in +local development. The upload URL's host is not the app host, so a network +that limits outbound requests must allow it. + Errors: a send that is not queued raises :class:`EmailError`. Branch on ``code``. The ones a setup usually hits: @@ -71,6 +84,27 @@ - ``client_disabled``: the client is disabled, so it sends nothing. - ``client_closed``: the client is shut down. +A send with attachments can also raise: + +- ``too_many_attachments`` or ``message_too_large``: the email is over the + limits above. The client raises these before any request. +- ``invalid_attachment``: the client can't send a file, such as one without + content. +- ``attachment_upload_failed``: PostHog's file storage refused an upload or + answered with a redirect, which the client never follows. ``message`` and + ``why`` carry its answer, and ``field`` names the attachment. +- ``network_error`` with ``field`` naming an attachment: the file storage + didn't answer, such as when a firewall blocks its host. +- ``api_error`` with ``field`` naming an attachment: PostHog named an upload + URL that isn't https. Fix the file storage's URL. Nothing was uploaded. +- ``upload_quota_exceeded`` (429): the project uploaded its daily limit of + file bytes. Files it already sent don't count. +- ``field_not_accepted`` with ``field`` ``attachments``: the project can't send + files yet. +- ``not_found`` (404) with ``field`` ``attachments``: this PostHog version + doesn't take files yet, or sending email from code is off for the project. + ``message`` carries PostHog's answer, and no email was sent. + Status: ``emails.get(message_id)`` reads what happened to a sent email, for 30 days after the send, as an :class:`EmailStatus`. A key with only ``transactional_email:read`` can read statuses but not send. ``get`` raises @@ -89,6 +123,9 @@ from __future__ import annotations import asyncio +import base64 +import binascii +import hashlib import importlib import json import logging @@ -102,6 +139,7 @@ from types import TracebackType from typing import ( Any, + Awaitable, Callable, Mapping, Optional, @@ -129,6 +167,7 @@ __all__ = [ "AsyncEmails", "CodeEmailArgs", + "EmailAttachment", "EmailError", "EmailStatus", "EmailTracking", @@ -141,6 +180,7 @@ log = logging.getLogger("posthog") _SEND_PATH = "/api/projects/@current/messaging/send/email/" +_UPLOADS_PATH = "/api/projects/@current/messaging/attachments/uploads/" _STATUS_PATH = "/api/projects/@current/messaging/emails/{message_id}/" _TEMPLATE_KEY = re.compile(r"[a-zA-Z0-9_-]{1,400}") _MAX_RETRY_DELAY = 30.0 @@ -148,6 +188,15 @@ _HEADER_TOKEN = re.compile(r"[\x21-\x7e]+") _IN_FLIGHT_CODES = frozenset({"idempotency_key_in_flight"}) _QUOTA_CODES = frozenset({"quota_limited"}) +_DAILY_QUOTA_CODES = frozenset({"upload_quota_exceeded"}) +_MAX_ATTACHMENTS = 10 +_STORAGE_RETRY_STATUSES = frozenset({429, 500, 502, 503, 504}) +_MAX_MESSAGE_BYTES = 40_000_000 +_HEADERS_AND_MIME_BOUNDARIES_BYTES = 64 * 1024 +_SUBJECT_BYTES_PER_ENCODED_WORD = 42 +_ENCODED_WORD_BYTES = 75 +_MESSAGE_LIMIT = "40 MB after Base64 encoding, about 29 MB of files" +_STORAGE_ERROR = re.compile(r"<(Code|Message)>([^<]{1,500})") class EmailError(Exception): @@ -217,6 +266,26 @@ class SendEmailResult: warnings: tuple[EmailWarning, ...] = () +class EmailAttachment(TypedDict): + """A file sent with the email. + + Args: + filename: The file name the recipient sees, such as ``invite.ics``. + content: The file as ``bytes``, ``bytearray`` or ``memoryview``, or + Base64 encoded as a string on one line with its padding, as + ``base64.b64encode`` gives. PostHog can't fetch a file from a path + or URL. + content_type: The MIME type. Without it, the type follows the file name. + content_id: Shows the file inline: reference it in ``html`` as + ```` with ``content_id`` ``logo``. + """ + + filename: str + content: Union[bytes, bytearray, memoryview, str] + content_type: NotRequired[str] + content_id: NotRequired[str] + + class _RecipientArgs(TypedDict): to: str distinct_id: NotRequired[Optional[str]] @@ -272,6 +341,10 @@ class CodeEmailArgs(_RecipientArgs): An optional field passed as ``None`` counts as left out. + A PostHog version that doesn't take a field yet, such as ``cc``, refuses + the send with an error that names the field in ``EmailError.field``. It + never drops the field and sends the rest. + Args: to: One recipient email address. template_key: Names the email, so its metrics, logs and status group @@ -283,6 +356,16 @@ class CodeEmailArgs(_RecipientArgs): without a plain-text part. from_: A verified sender, such as ``"Acme "``. reply_to: One address or a list of addresses. An empty list sends no reply-to. + cc: One address or a list of addresses that get a visible copy, at + most 10. + bcc: One address or a list of addresses that get a hidden copy, at + most 10. + attachments: Files sent with the email: at most 10 files and 40 MB + per message after Base64 encoding, about 29 MB of files. The + client checks both before any request. It uploads each file that + PostHog doesn't hold yet, then sends the email. A file the client + can't send raises :class:`EmailError` with ``code`` + ``invalid_attachment``. capture_engagement: Deprecated, use ``tracking={"engagement": False}``. It keeps working, but PostHog refuses a send that sets both ``capture_engagement`` and ``tracking``. Captures the email's @@ -324,17 +407,24 @@ class CodeEmailArgs(_RecipientArgs): text: NotRequired[Optional[str]] from_: NotRequired[Optional[str]] reply_to: NotRequired[Optional[Union[str, Sequence[str]]]] + cc: NotRequired[Optional[Union[str, Sequence[str]]]] + bcc: NotRequired[Optional[Union[str, Sequence[str]]]] + attachments: NotRequired[Optional[Sequence[EmailAttachment]]] capture_engagement: NotRequired[Optional[bool]] sensitive: NotRequired[Optional[bool]] tracking: NotRequired[Optional[Union[bool, EmailTracking]]] +_ATTACHMENT_FIELDS = frozenset(EmailAttachment.__annotations__) _OPTIONAL_CODE_FIELDS = frozenset( { "html", "text", "from_", "reply_to", + "cc", + "bcc", + "attachments", "capture_engagement", "sensitive", "tracking", @@ -375,12 +465,32 @@ class EmailStatus: reason: Optional[str] = None +@dataclass(frozen=True) +class _File: + sha256: str + content: bytes + first_attachment_index: int + + @property + def size(self) -> int: + return len(self.content) + + +@dataclass(frozen=True) +class _Upload: + file: _File + url: str + method: str + headers: dict[str, str] + + @dataclass(frozen=True) class _PreparedSend: path: str body: bytes idempotency_key: str max_retries: Optional[int] = None + files: tuple[_File, ...] = () def _prepare(args: Mapping[str, Any]) -> _PreparedSend: @@ -434,15 +544,20 @@ def _prepare_code_send(args: Mapping[str, Any]) -> _PreparedSend: if not (content.get("html") or content.get("text")): raise ValueError("html or text must be a nonempty string.") + references, files = _attachments(args.get("attachments", []), content) + payload: dict[str, Any] = {"to": to, "template_key": template_key, **content} if distinct_id is not None: payload["distinct_id"] = distinct_id payload.update(_sender(args)) + if references: + payload["attachments"] = references return _PreparedSend( path=_SEND_PATH, body=_encode(payload), idempotency_key=idempotency_key, max_retries=max_retries, + files=_each_file_once(files), ) @@ -450,9 +565,10 @@ def _sender(args: Mapping[str, Any]) -> dict[str, Any]: sender: dict[str, Any] = {} if "from_" in args: sender["from"] = _nonempty_text(args, "from_") - reply_to = _addresses("reply_to", args["reply_to"]) if "reply_to" in args else [] - if reply_to: - sender["reply_to"] = reply_to + for field in ("reply_to", "cc", "bcc"): + addresses = _addresses(field, args[field]) if field in args else [] + if addresses: + sender[field] = addresses for flag in ("capture_engagement", "sensitive"): if flag in args: if not isinstance(args[flag], bool): @@ -484,6 +600,189 @@ def _addresses(field: str, value: Any) -> list[str]: return list(addresses) +def _each_file_once(files: Sequence[_File]) -> tuple[_File, ...]: + by_hash: dict[str, _File] = {} + for file in files: + by_hash.setdefault(file.sha256, file) + return tuple(by_hash.values()) + + +def _invalid_attachment(message: str, field: str) -> EmailError: + return EmailError( + name="validation_error", + code="invalid_attachment", + message=message, + field=field, + ) + + +def _attachments( + attachments: Any, content: Mapping[str, str] +) -> tuple[list[dict[str, str]], tuple[_File, ...]]: + if isinstance( + attachments, (str, bytes, bytearray, memoryview, Mapping) + ) or not isinstance(attachments, Sequence): + raise _invalid_attachment("attachments must be a list of files.", "attachments") + if len(attachments) > _MAX_ATTACHMENTS: + raise _over_the_limit( + "too_many_attachments", + f"An email carries at most {_MAX_ATTACHMENTS} files, so no email was sent.", + fix=f"Send at most {_MAX_ATTACHMENTS} files, and link to any others " + "from the email body.", + ) + checked = [ + _checked_attachment(index, file) for index, file in enumerate(attachments) + ] + _check_message_size( + [ + _content_size(f"attachments__{index}", file.get("content")) + for index, file in enumerate(checked) + ], + content, + ) + prepared = [_attachment(index, file) for index, file in enumerate(checked)] + return [reference for reference, _ in prepared], tuple(file for _, file in prepared) + + +def _checked_attachment(index: int, file: Any) -> Mapping[str, Any]: + field = f"attachments__{index}" + if not isinstance(file, Mapping): + raise _invalid_attachment( + "An attachment is a mapping with filename and content. " + "Pass the file's bytes as its content.", + field, + ) + if "path" in file: + raise _invalid_attachment( + "PostHog can't fetch an attachment from a path or URL. " + "Read the file and pass its bytes as content instead.", + f"{field}__path", + ) + unknown = [name for name in file if name not in _ATTACHMENT_FIELDS] + if unknown: + raise _invalid_attachment( + "An attachment takes filename, content, content_type and content_id, " + f"not {unknown[0]}.", + f"{field}__{unknown[0]}", + ) + if not isinstance(file.get("filename"), str) or not file["filename"]: + raise _invalid_attachment( + "An attachment needs a filename, such as 'invoice.pdf'.", + f"{field}__filename", + ) + for name in ("content_type", "content_id"): + if name in file and not isinstance(file[name], str): + raise _invalid_attachment( + f"An attachment's {name} must be a string.", f"{field}__{name}" + ) + return file + + +def _content_size(field: str, content: Any) -> int: + if isinstance(content, (bytes, bytearray, memoryview)): + size = memoryview(content).nbytes + elif isinstance(content, str): + data_end = len(content) + while data_end and content[data_end - 1] == "=": + data_end -= 1 + if len(content) - data_end > 2: + raise _not_standard_base64(field) + size = data_end * 3 // 4 + else: + raise _invalid_attachment( + "An attachment needs content: bytes or a Base64 string. " + "Convert a list of byte values with bytes(content).", + f"{field}__content", + ) + if not size: + raise _invalid_attachment( + "An attachment's content is empty, and an empty file gives the " + "recipient nothing to open. Leave the file out.", + f"{field}__content", + ) + return size + + +def _attachment(index: int, file: Mapping[str, Any]) -> tuple[dict[str, str], _File]: + content = _file_bytes(f"attachments__{index}", file["content"]) + sha256 = hashlib.sha256(content).hexdigest() + reference = {"sha256": sha256, "filename": file["filename"]} + for name in ("content_type", "content_id"): + if name in file: + reference[name] = file[name] + return reference, _File( + sha256=sha256, content=content, first_attachment_index=index + ) + + +def _file_bytes(field: str, content: Union[bytes, bytearray, memoryview, str]) -> bytes: + if not isinstance(content, str): + return bytes(content) + try: + return base64.b64decode(content, validate=True) + except (binascii.Error, ValueError): + raise _not_standard_base64(field) from None + + +def _not_standard_base64(field: str) -> EmailError: + return _invalid_attachment( + "An attachment's content string must be standard Base64 on one line " + "with its padding, as base64.b64encode gives. Pass the file's bytes " + "instead, such as text.encode() for text.", + f"{field}__content", + ) + + +def _check_message_size(file_sizes: Sequence[int], content: Mapping[str, str]) -> None: + if ( + file_sizes + and _estimated_message_bytes(file_sizes, content) > _MAX_MESSAGE_BYTES + ): + raise _over_the_limit( + "message_too_large", + f"These files make the email larger than {_MESSAGE_LIMIT}, so no email was sent.", + fix="Send fewer or smaller files, or link to large files from the email body.", + ) + + +def _over_the_limit(code: str, message: str, *, fix: str) -> EmailError: + return EmailError( + name="validation_error", + code=code, + message=message, + why=f"An email carries at most {_MAX_ATTACHMENTS} files and {_MESSAGE_LIMIT}.", + fix=fix, + field="attachments", + ) + + +def _estimated_message_bytes( + file_sizes: Sequence[int], content: Mapping[str, str] +) -> int: + html, text = content.get("html") or "", content.get("text") or "" + body_sizes = [len(body.encode("utf-8")) for body in (html, text or html) if body] + return ( + _HEADERS_AND_MIME_BOUNDARIES_BYTES + + sum(_base64_transfer_bytes(size) for size in [*file_sizes, *body_sizes]) + + _encoded_subject_bytes(content["subject"]) + ) + + +def _encoded_subject_bytes(subject: str) -> int: + return ( + _ceil_div(len(subject.encode("utf-8")), _SUBJECT_BYTES_PER_ENCODED_WORD) + * _ENCODED_WORD_BYTES + ) + + +def _base64_transfer_bytes(size: int) -> int: + return _ceil_div(_ceil_div(size, 3) * 4 * 78, 76) + + +def _ceil_div(dividend: int, divisor: int) -> int: + return -(-dividend // divisor) + + def _string(args: Mapping[str, Any], name: str) -> str: value = args[name] if not isinstance(value, str): @@ -633,6 +932,8 @@ def _error_name(status: Optional[int], code: str) -> str: else "validation_error" ) if status == 429: + if code in _DAILY_QUOTA_CODES: + return "daily_quota_exceeded" return ( "monthly_quota_exceeded" if code in _QUOTA_CODES else "rate_limit_exceeded" ) @@ -654,10 +955,18 @@ def _retry_after(header: Optional[str]) -> Optional[float]: return max(0.0, seconds) +def _json_object(body: bytes) -> dict[str, Any]: + try: + payload = json.loads(body) + except ValueError: + return {} + return payload if isinstance(payload, dict) else {} + + def _parse_response( - status: int, payload: Any, retry_after_header: Optional[str] + status: int, body: bytes, retry_after_header: Optional[str] ) -> SendEmailResult: - fields = payload if isinstance(payload, dict) else {} + fields = _json_object(body) if 200 <= status < 300: message_id = _text(fields.get("message_id")) if message_id is None: @@ -697,9 +1006,9 @@ def _timestamp(value: Any) -> Optional[datetime]: def _parse_status_response( - status: int, payload: Any, retry_after_header: Optional[str] + status: int, body: bytes, retry_after_header: Optional[str] ) -> EmailStatus: - fields = payload if isinstance(payload, dict) else {} + fields = _json_object(body) if not 200 <= status < 300: raise _response_error(fields, status, retry_after_header) message_id = _text(fields.get("id")) @@ -724,6 +1033,135 @@ def _parse_status_response( ) +def _pending_uploads_parser( + files: Sequence[_File], plain_http_allowed: bool +) -> _Parse[list[_Upload]]: + def parse( + status: int, body: bytes, retry_after_header: Optional[str] + ) -> list[_Upload]: + fields = _json_object(body) + if status == 404: + raise _upload_route_not_found(fields, retry_after_header) + if not 200 <= status < 300: + attr = _attachments_field(fields.get("attr"), files) + raise _response_error({**fields, "attr": attr}, status, retry_after_header) + slots = fields.get("files") + if not isinstance(slots, list) or len(slots) != len(files): + raise _no_upload_slots(status) + pending = ( + _pending_upload(slot, file, status) for slot, file in zip(slots, files) + ) + uploads = [upload for upload in pending if upload is not None] + for upload in uploads: + if not plain_http_allowed and not upload.url.lower().startswith("https://"): + raise _upload_url_not_https(upload.file.first_attachment_index, status) + return uploads + + return parse + + +def _attachments_field(attr: Any, files: Sequence[_File]) -> Any: + if not isinstance(attr, str) or not attr.startswith("files"): + return attr + index = attr.split("__")[1] if attr.startswith("files__") else "" + if index.isdigit() and int(index) < len(files): + return f"attachments__{files[int(index)].first_attachment_index}" + return "attachments" + + +def _pending_upload(slot: Any, file: _File, status: int) -> Optional[_Upload]: + if not isinstance(slot, dict) or slot.get("sha256") != file.sha256: + raise _no_upload_slots(status) + if slot.get("status") == "present": + return None + url, method, headers = ( + _text(slot.get("url")), + _text(slot.get("method", "PUT")), + slot.get("headers"), + ) + if ( + slot.get("status") != "upload" + or url is None + or method is None + or not isinstance(headers, dict) + or not all(isinstance(value, str) for value in headers.values()) + ): + raise _no_upload_slots(status) + return _Upload(file=file, url=url, method=method, headers=headers) + + +def _upload_route_not_found( + fields: Mapping[str, Any], retry_after_header: Optional[str] +) -> EmailError: + unanswered = { + "code": "not_found", + "why": "This PostHog version doesn't take attachments yet, or sending " + "email from code is off for the project.", + "fix": "Send the email without attachments, or check that the project " + "can send email from code.", + } + answered = {key: value for key, value in fields.items() if value is not None} + return _response_error( + {**unanswered, **answered, "attr": "attachments"}, 404, retry_after_header + ) + + +def _upload_url_not_https(index: int, status: int) -> EmailError: + return EmailError( + name=_error_name(status, "api_error"), + code="api_error", + message=f"PostHog named an upload URL that isn't https, so attachments[{index}] " + "would travel unencrypted. No email was sent.", + status_code=status, + why="PostHog's file storage is set up with a plain HTTP URL.", + fix="Fix the file storage's URL so it uses https. For local development " + "over plain HTTP, give the client a host that starts with http://.", + field=f"attachments__{index}", + ) + + +def _no_upload_slots(status: int) -> EmailError: + return _sdk_error( + "api_error", "PostHog answered without an upload slot for each file.", status + ) + + +def _upload_parser(index: int) -> _Parse[None]: + def parse(status: int, body: bytes, retry_after_header: Optional[str]) -> None: + if not 200 <= status < 300: + raise _upload_error(index, status, body, retry_after_header) + + return parse + + +def _upload_error( + index: int, status: int, body: bytes, retry_after_header: Optional[str] +) -> EmailError: + answer = dict(_STORAGE_ERROR.findall(body.decode("utf-8", "replace"))) + storage_answer = " ".join(filter(None, (f"HTTP {status}", answer.get("Code")))) + return EmailError( + name=_storage_error_name(status), + code="attachment_upload_failed", + message=f"PostHog's file storage refused the upload of attachments[{index}] " + f"with {storage_answer}, so no email was sent.", + status_code=status, + why=answer.get("Message"), + fix="Send the email again, which asks for a fresh upload URL. If the " + "upload keeps failing, check whether a proxy between your app and " + "PostHog's file storage changes or holds up uploads.", + field=f"attachments__{index}", + retry_after=_retry_after(retry_after_header), + ) + + +def _storage_error_name(status: int) -> str: + if status == 429: + return "rate_limit_exceeded" + if status in _STORAGE_RETRY_STATUSES: + return "service_unavailable" + return "application_error" + + def _response_error( fields: Mapping[str, Any], status: int, retry_after_header: Optional[str] ) -> EmailError: @@ -745,10 +1183,12 @@ def _is_retryable(error: EmailError) -> bool: # A send carries an idempotency key, so a retry can't queue a second email. A status read changes nothing. if error.status_code is None: return error.code == "network_error" + if error.code == "attachment_upload_failed": + return error.status_code in _STORAGE_RETRY_STATUSES if error.status_code == 409: return error.code in _IN_FLIGHT_CODES if error.status_code == 429: - return error.code not in _QUOTA_CODES + return error.code not in _QUOTA_CODES | _DAILY_QUOTA_CODES return error.status_code == 503 @@ -800,7 +1240,7 @@ def _headers( "transactional_email:read or transactional_email:write scope." ) _ResultT = TypeVar("_ResultT") -_Parse = Callable[[int, Any, Optional[str]], _ResultT] +_Parse = Callable[[int, bytes, Optional[str]], _ResultT] class _BearerAuth(requests.auth.AuthBase): @@ -813,6 +1253,24 @@ def __call__(self, request: requests.PreparedRequest) -> requests.PreparedReques return request +def _storage_unreachable(index: int, exc: Exception) -> EmailError: + return EmailError( + name="service_unavailable", + code="network_error", + message=f"PostHog's file storage didn't answer the upload of attachments[{index}] " + f"({type(exc).__name__}), so no email was sent.", + fix="Send the email again. If it keeps failing, check that your network " + "lets the app reach PostHog's file storage.", + field=f"attachments__{index}", + ) + + +def _files_body(files: Sequence[_File]) -> bytes: + return _encode( + {"files": [{"sha256": file.sha256, "size": file.size} for file in files]} + ) + + class _EmailsBase: def __init__( self, @@ -935,6 +1393,11 @@ def send(self, **kwargs: Any) -> SendEmailResult: Raises: EmailError: PostHog did not queue the email, or the client could not tell whether it did. + With attachments, ``code`` is ``too_many_attachments``, + ``message_too_large`` or ``invalid_attachment`` before any + request, ``attachment_upload_failed`` when the storage + refused an upload, and ``network_error`` with ``field`` + naming the attachment when the storage didn't answer. TypeError, ValueError: An argument has the wrong type or format. Examples: @@ -965,6 +1428,8 @@ def send(self, **kwargs: Any) -> SendEmailResult: return result def _deliver(self, prepared: _PreparedSend, secret_key: str) -> SendEmailResult: + if prepared.files: + self._upload(prepared, secret_key) headers = _headers( secret_key, self.migrated_from, @@ -972,14 +1437,29 @@ def _deliver(self, prepared: _PreparedSend, secret_key: str) -> SendEmailResult: idempotency_key=prepared.idempotency_key, ) return self._call( - "POST", - prepared.path, - prepared.body, - headers, + self._posthog_request("POST", prepared.path, prepared.body, headers), _parse_response, prepared.max_retries, ) + def _upload(self, prepared: _PreparedSend, secret_key: str) -> None: + headers = _headers(secret_key, self.migrated_from, has_body=True) + uploads = self._call( + self._posthog_request( + "POST", _UPLOADS_PATH, _files_body(prepared.files), headers + ), + _pending_uploads_parser( + prepared.files, self.host.lower().startswith("http://") + ), + prepared.max_retries, + ) + for upload in uploads: + self._call( + self._storage_request(upload), + _upload_parser(upload.file.first_attachment_index), + prepared.max_retries, + ) + def get(self, message_id: Union[str, UUID]) -> EmailStatus: """Read what happened to an email that :meth:`send` queued. @@ -1008,41 +1488,61 @@ def get(self, message_id: Union[str, UUID]) -> EmailStatus: finish = self._start(_GET_NOT_DONE) try: headers = _headers(self._secret_key(_GET_KEY_NEEDED), self.migrated_from) - return self._call("GET", path, None, headers, _parse_status_response) + return self._call( + self._posthog_request("GET", path, None, headers), + _parse_status_response, + ) finally: finish() + def _posthog_request( + self, method: str, path: str, body: Optional[bytes], headers: dict[str, str] + ) -> Callable[[], requests.Response]: + session, auth = self._get_session(), _BearerAuth(headers) + return lambda: session.request( + method, + self.host + path, + data=body, + auth=auth, + timeout=self.timeout, + allow_redirects=False, + ) + + def _storage_request(self, upload: _Upload) -> Callable[[], requests.Response]: + session = self._get_session() + request = requests.Request( + upload.method, upload.url, data=upload.file.content, headers=upload.headers + ).prepare() + transport = session.merge_environment_settings( + upload.url, {}, False, None, None + ) + + def send() -> requests.Response: + try: + return session.send( + request, timeout=self.timeout, allow_redirects=False, **transport + ) + except requests.exceptions.RequestException as exc: + raise _storage_unreachable( + upload.file.first_attachment_index, exc + ) from None + + return send + def _call( self, - method: str, - path: str, - body: Optional[bytes], - headers: dict[str, str], + send: Callable[[], requests.Response], parse: _Parse[_ResultT], max_retries: Optional[int] = None, ) -> _ResultT: - auth = _BearerAuth(headers) - session = self._get_session() - url = self.host + path retries = self.max_retries if max_retries is None else max_retries for attempt in range(retries + 1): try: - response = session.request( - method, - url, - data=body, - auth=auth, - timeout=self.timeout, - allow_redirects=False, - ) + response = send() try: - try: - payload = response.json() - except ValueError: - payload = None return parse( response.status_code, - payload, + response.content, response.headers.get("Retry-After"), ) finally: @@ -1149,6 +1649,8 @@ async def send(self, **kwargs: Any) -> SendEmailResult: async def _deliver( self, prepared: _PreparedSend, secret_key: str ) -> SendEmailResult: + if prepared.files: + await self._upload(prepared, secret_key) headers = _headers( secret_key, self.migrated_from, @@ -1156,14 +1658,29 @@ async def _deliver( idempotency_key=prepared.idempotency_key, ) return await self._call( - "POST", - prepared.path, - prepared.body, - headers, + self._posthog_request("POST", prepared.path, prepared.body, headers), _parse_response, prepared.max_retries, ) + async def _upload(self, prepared: _PreparedSend, secret_key: str) -> None: + headers = _headers(secret_key, self.migrated_from, has_body=True) + uploads = await self._call( + self._posthog_request( + "POST", _UPLOADS_PATH, _files_body(prepared.files), headers + ), + _pending_uploads_parser( + prepared.files, self.host.lower().startswith("http://") + ), + prepared.max_retries, + ) + for upload in uploads: + await self._call( + self._storage_request(upload), + _upload_parser(upload.file.first_attachment_index), + prepared.max_retries, + ) + async def get(self, message_id: Union[str, UUID]) -> EmailStatus: """Read what happened to an email that :meth:`send` queued. @@ -1174,42 +1691,63 @@ async def get(self, message_id: Union[str, UUID]) -> EmailStatus: finish = self._start(_GET_NOT_DONE) try: headers = _headers(self._secret_key(_GET_KEY_NEEDED), self.migrated_from) - return await self._call("GET", path, None, headers, _parse_status_response) + return await self._call( + self._posthog_request("GET", path, None, headers), + _parse_status_response, + ) finally: finish() + def _posthog_request( + self, method: str, path: str, body: Optional[bytes], headers: dict[str, str] + ) -> Callable[[], Awaitable[Any]]: + _, client = self._get_client() + # An injected client may carry its own auth and timeout, so set both per request: auth=None keeps + # the secret key's Bearer header, and the request timeout applies whatever the client's default. + return lambda: client.request( + method, + self.host + path, + content=body, + headers=headers, + auth=None, + timeout=self.timeout, + ) + + def _storage_request(self, upload: _Upload) -> Callable[[], Awaitable[Any]]: + httpx, client = self._get_client() + request = httpx.Request( + upload.method, + upload.url, + content=upload.file.content, + headers=upload.headers, + extensions={"timeout": httpx.Timeout(self.timeout).as_dict()}, + ) + + async def send() -> Any: + try: + return await client.send(request, auth=None, follow_redirects=False) + except httpx.RequestError as exc: + raise _storage_unreachable( + upload.file.first_attachment_index, exc + ) from None + + return send + async def _call( self, - method: str, - path: str, - body: Optional[bytes], - headers: dict[str, str], + send: Callable[[], Awaitable[Any]], parse: _Parse[_ResultT], max_retries: Optional[int] = None, ) -> _ResultT: - httpx, client = self._get_client() - url = self.host + path + httpx, _ = self._get_client() retries = self.max_retries if max_retries is None else max_retries for attempt in range(retries + 1): try: - # An injected client may carry its own auth and timeout, so set both per request: auth=None keeps - # the secret key's Bearer header, and the request timeout applies whatever the client's default. - response = await client.request( - method, - url, - content=body, - headers=headers, - auth=None, - timeout=self.timeout, - ) + response = await send() try: - try: - payload = response.json() - except ValueError: - payload = None return parse( response.status_code, - payload, + response.content, response.headers.get("Retry-After"), ) finally: diff --git a/posthog/test/test_emails.py b/posthog/test/test_emails.py index b13b8a0c6..82b8a8dce 100644 --- a/posthog/test/test_emails.py +++ b/posthog/test/test_emails.py @@ -1,14 +1,28 @@ from __future__ import annotations import asyncio +import base64 +import hashlib +import hmac import json import logging +import re +import ssl import threading -from datetime import datetime, timezone +import time +import traceback +import tracemalloc +from contextlib import contextmanager +from dataclasses import dataclass, field, replace +from datetime import datetime, timedelta, timezone from decimal import Decimal -from http.server import BaseHTTPRequestHandler, HTTPServer -from typing import Any, Iterator, Optional, Union +from http.server import BaseHTTPRequestHandler, HTTPServer, ThreadingHTTPServer +from ipaddress import ip_address +from pathlib import Path +from typing import Any, Callable, Iterator, Mapping, Optional, Union from unittest import mock +from urllib.parse import parse_qsl, urlencode, urlsplit +from uuid import uuid4 import httpx import pytest @@ -622,6 +636,8 @@ def handler(request: httpx.Request) -> httpx.Response: HTML = "

Reset it here.

" +INVITE = b"BEGIN:VCALENDAR\r\nEND:VCALENDAR\r\n" +INVITE_BASE64 = "QkVHSU46VkNBTEVOREFSDQpFTkQ6VkNBTEVOREFSDQo=" def send_code(emails: Emails, **overrides: Any): @@ -845,12 +861,13 @@ def handler(request: httpx.Request) -> httpx.Response: ("empty_html_and_text", {"html": "", "text": ""}, ValueError), ("blank_reply_to", {"html": HTML, "reply_to": " "}, ValueError), ("reply_to_not_text", {"html": HTML, "reply_to": 123}, TypeError), + ("blank_cc", {"html": HTML, "cc": " "}, ValueError), + ("bcc_not_text", {"html": HTML, "bcc": [123]}, TypeError), ( "template_data_on_code_form", {"html": HTML, "template_data": {"first_name": "Jane"}}, TypeError, ), - ("cc_on_code_form", {"html": HTML, "cc": "a@example.com"}, TypeError), ( "capture_engagement_on_template_form", { @@ -861,6 +878,36 @@ def handler(request: httpx.Request) -> httpx.Response: }, TypeError, ), + ( + "cc_on_template_form", + { + "template_id": TEMPLATE_ID, + "template_key": None, + "subject": None, + "cc": "a@example.com", + }, + TypeError, + ), + ( + "bcc_on_template_form", + { + "template_id": TEMPLATE_ID, + "template_key": None, + "subject": None, + "bcc": ["a@example.com"], + }, + TypeError, + ), + ( + "attachments_on_template_form", + { + "template_id": TEMPLATE_ID, + "template_key": None, + "subject": None, + "attachments": [{"filename": "a.txt", "content": b"hi"}], + }, + TypeError, + ), ( "tracking_on_template_form", { @@ -900,3 +947,1737 @@ def test_rejects_a_code_send_that_mixes_or_misses_fields(_name, overrides, error ) assert adapter.requests == [] + + +def test_a_refused_send_names_the_field_from_attr(): + posthog = FakePostHog() + posthog.send_fields = SEND_FIELDS - {"cc"} + + with pytest.raises(EmailError) as raised: + send_code(emails_on(posthog), html=HTML, cc="manager@example.com") + + assert (raised.value.code, raised.value.field) == ("invalid_input", "cc") + assert posthog.delivered == [] + + +POSTHOG_HOST = "https://us.posthog.com" +UPLOADS_PATH = "/api/projects/@current/messaging/attachments/uploads/" +SEND_PATH = "/api/projects/@current/messaging/send/email/" +STORAGE_URL = "https://files.example.com/team/1" +STORAGE_SIGNING_KEY = b"fake-storage-signing-key" +URL_LIFETIME_SECONDS = 900 +SEND_FIELDS = frozenset( + { + "to", + "template_key", + "subject", + "html", + "text", + "from", + "reply_to", + "cc", + "bcc", + "capture_engagement", + "sensitive", + "tracking", + "distinct_id", + "attachments", + } +) +REFERENCE_FIELDS = frozenset({"sha256", "filename", "content_type", "content_id"}) +SHA256_HEX = re.compile(r"[0-9a-f]{64}") + + +@dataclass(frozen=True) +class Answer: + status: int + body: bytes = b"" + headers: dict[str, str] = field(default_factory=dict) + + @classmethod + def of_json(cls, status: int, payload: dict[str, Any]) -> Answer: + return cls( + status, json.dumps(payload).encode(), {"Content-Type": "application/json"} + ) + + +def api_error( + status: int, code: str, detail: str, kind: str = "validation_error", **extra: Any +) -> Answer: + return Answer.of_json( + status, {"type": kind, "code": code, "detail": detail, **extra} + ) + + +def storage_error(status: int, code: str, message: str) -> Answer: + return Answer( + status, + f'{code}' + f"{message}".encode(), + {"Content-Type": "application/xml"}, + ) + + +UNAUTHORIZED = api_error( + 401, + "authentication_failed", + "Incorrect authentication credentials.", + kind="authentication_error", +) + + +@dataclass(frozen=True) +class Incoming: + method: str + url: str + headers: Mapping[str, str] + body: bytes + + @property + def path(self) -> str: + return urlsplit(self.url).path + + @property + def is_authorized(self) -> bool: + return self.headers.get("authorization") == f"Bearer {SECRET_KEY}" + + +class Unreachable(Exception): + pass + + +Handler = Callable[[Incoming], Answer] +Fault = Callable[[Incoming, Handler], Answer] + + +def answers(answer: Answer) -> Fault: + return lambda _incoming, _handle: answer + + +def loses_the_reply(answer: Answer) -> Fault: + def fault(incoming: Incoming, handle: Handler) -> Answer: + handle(incoming) + return answer + + return fault + + +def unreachable(incoming: Incoming, _handle: Handler) -> Answer: + raise Unreachable(f"Max retries exceeded with url: {incoming.url}") + + +def alters_the_bytes(incoming: Incoming, handle: Handler) -> Answer: + return handle(replace(incoming, body=incoming.body[::-1])) + + +@dataclass(frozen=True) +class DeliveredFile: + filename: str + content: bytes + content_type: Optional[str] = None + content_id: Optional[str] = None + + +@dataclass(frozen=True) +class DeliveredEmail: + message_id: str + to: str + cc: tuple[str, ...] + bcc: tuple[str, ...] + subject: str + html: Optional[str] + text: Optional[str] + attachments: tuple[DeliveredFile, ...] + + +def checksum(content: bytes) -> str: + return base64.b64encode(hashlib.sha256(content).digest()).decode() + + +def storage_signature( + method: str, path: str, query: Mapping[str, str], headers: Mapping[str, str] +) -> str: + canonical = "\n".join( + [ + method, + path, + urlencode(sorted(query.items())), + headers.get("content-length", ""), + headers.get("x-amz-checksum-sha256", ""), + ] + ) + return hmac.new(STORAGE_SIGNING_KEY, canonical.encode(), hashlib.sha256).hexdigest() + + +def json_or_none(body: bytes) -> Any: + try: + return json.loads(body) + except ValueError: + return None + + +def files_asked_for(body: bytes) -> Optional[list[tuple[str, int]]]: + payload = json_or_none(body) + if not isinstance(payload, dict) or set(payload) != {"files"}: + return None + files = payload["files"] + if not isinstance(files, list) or not files: + return None + if not all( + isinstance(file, dict) + and set(file) == {"sha256", "size"} + and isinstance(file["sha256"], str) + and SHA256_HEX.fullmatch(file["sha256"]) + and type(file["size"]) is int + and file["size"] > 0 + for file in files + ): + return None + return [(file["sha256"], file["size"]) for file in files] + + +class FakePostHog: + def __init__( + self, storage_url: str = STORAGE_URL, posthog_host: str = POSTHOG_HOST + ) -> None: + self.storage_url = storage_url + self.posthog_host = posthog_host + self._bucket: dict[str, bytes] = {} + self._content_types: dict[str, Optional[str]] = {} + self.writes = 0 + self.delivered: list[DeliveredEmail] = [] + self.project_can_send_files = True + self.send_fields = SEND_FIELDS + self._sent_by_idempotency_key: dict[str, str] = {} + self._faults: dict[str, list[Fault]] = { + "uploads": [], + "send": [], + "storage": [], + } + self._signing_delay = 0.0 + + @property + def stored_files(self) -> list[bytes]: + return sorted(self._bucket.values()) + + @property + def stored_content_types(self) -> list[Optional[str]]: + return list(self._content_types.values()) + + def already_has(self, content: bytes) -> None: + self._bucket[hashlib.sha256(content).hexdigest()] = content + + def inject(self, route: str, *faults: Fault) -> None: + self._faults[route].extend(faults) + + def sign_urls_in_the_past(self, minutes: int) -> None: + self._signing_delay = minutes * 60.0 + + def handle(self, incoming: Incoming) -> Answer: + route, handler = self._route(incoming) + faults = self._faults.get(route) + if faults: + return faults.pop(0)(incoming, handler) + return handler(incoming) + + def _route(self, incoming: Incoming) -> tuple[str, Handler]: + if incoming.url.startswith(self.storage_url + "/"): + if incoming.method != "PUT": + return "storage", lambda _incoming: storage_error( + 405, + "MethodNotAllowed", + "The specified method is not allowed against this resource.", + ) + return "storage", self._store + if incoming.url.startswith(self.posthog_host + "/") and incoming.path in ( + UPLOADS_PATH, + SEND_PATH, + ): + route = "uploads" if incoming.path == UPLOADS_PATH else "send" + if incoming.method != "POST": + return route, lambda _incoming: api_error( + 405, + "method_not_allowed", + f'Method "{incoming.method}" not allowed.', + kind="invalid_request", + ) + return route, (self._answer_uploads if route == "uploads" else self._send) + return "unknown", lambda _incoming: api_error( + 404, "not_found", "Not found.", kind="invalid_request" + ) + + def _answer_uploads(self, incoming: Incoming) -> Answer: + if not incoming.is_authorized: + return UNAUTHORIZED + if not self.project_can_send_files: + return api_error( + 400, + "field_not_accepted", + "This project can't send files yet, so PostHog didn't prepare the uploads.", + kind="invalid_request", + attr="files", + why="Sending files isn't available to this project yet.", + fix="Send the email without files.", + ) + files = files_asked_for(incoming.body) + if files is None: + return api_error( + 400, "invalid_input", "Name each file by sha256 and size.", attr="files" + ) + return Answer.of_json( + 200, {"files": [self._slot(sha256, size) for sha256, size in files]} + ) + + def _slot(self, sha256: str, size: int) -> dict[str, Any]: + if sha256 in self._bucket: + return { + "sha256": sha256, + "status": "present", + "expires_at": "2026-12-16T00:00:00Z", + } + url = f"{self.storage_url}/{sha256}" + headers = { + "content-length": str(size), + "x-amz-checksum-sha256": base64.b64encode(bytes.fromhex(sha256)).decode(), + } + query = { + "X-Amz-Date": str(int(time.time() - self._signing_delay)), + "X-Amz-Expires": str(URL_LIFETIME_SECONDS), + "X-Amz-SignedHeaders": "content-length;x-amz-checksum-sha256", + } + signature = storage_signature("PUT", urlsplit(url).path, query, headers) + return { + "sha256": sha256, + "status": "upload", + "url": f"{url}?{urlencode({**query, 'X-Amz-Signature': signature})}", + "method": "PUT", + "headers": headers, + "url_expires_at": "2026-11-09T12:15:00Z", + } + + def _store(self, incoming: Incoming) -> Answer: + if "authorization" in incoming.headers: + return storage_error( + 400, + "InvalidArgument", + "Only one auth mechanism allowed; only the X-Amz-Algorithm query " + "parameter, Signature query string parameter or the Authorization " + "header should be specified", + ) + query = dict(parse_qsl(urlsplit(incoming.url).query, keep_blank_values=True)) + signature = query.pop("X-Amz-Signature", "") + expected = storage_signature( + incoming.method, incoming.path, query, incoming.headers + ) + if not hmac.compare_digest(signature, expected): + return storage_error( + 403, + "SignatureDoesNotMatch", + "The request signature we calculated does not match the signature you provided.", + ) + if time.time() > int(query["X-Amz-Date"]) + int(query["X-Amz-Expires"]): + return storage_error(403, "AccessDenied", "Request has expired") + if incoming.headers["content-length"] != str( + len(incoming.body) + ) or incoming.headers["x-amz-checksum-sha256"] != checksum(incoming.body): + return storage_error( + 400, + "BadDigest", + "The SHA256 you specified did not match the calculated checksum.", + ) + key = incoming.path.rsplit("/", 1)[1] + self._bucket[key] = incoming.body + self._content_types[key] = incoming.headers.get("content-type") + self.writes += 1 + return Answer(200) + + def _send(self, incoming: Incoming) -> Answer: + if not incoming.is_authorized: + return UNAUTHORIZED + idempotency_key = incoming.headers.get("idempotency-key", "") + if idempotency_key in self._sent_by_idempotency_key: + return Answer.of_json( + 200, + { + "message_id": self._sent_by_idempotency_key[idempotency_key], + "replayed": True, + }, + ) + payload = json_or_none(incoming.body) + if not isinstance(payload, dict): + return api_error(400, "invalid_input", "Send a JSON object.") + unknown = sorted(set(payload) - self.send_fields) + if unknown: + return api_error( + 400, + "invalid_input", + f"PostHog doesn't take {unknown[0]}.", + attr=unknown[0], + ) + references = payload.get("attachments", []) + for index, reference in enumerate(references): + refused = self._refused_reference(index, reference) + if refused is not None: + return refused + message_id = str(uuid4()) + self.delivered.append(self._email(message_id, payload, references)) + if idempotency_key: + self._sent_by_idempotency_key[idempotency_key] = message_id + return Answer.of_json(202, {"message_id": message_id}) + + def _refused_reference(self, index: int, reference: Any) -> Optional[Answer]: + if ( + not isinstance(reference, dict) + or not {"sha256", "filename"} <= set(reference) <= REFERENCE_FIELDS + ): + return api_error( + 400, + "invalid_input", + "An attachment names sha256 and filename.", + attr=f"attachments__{index}", + ) + if reference["sha256"] not in self._bucket: + return api_error( + 400, + "attachment_not_uploaded", + "PostHog doesn't hold this file. Upload it first.", + attr=f"attachments__{index}", + ) + return None + + def _email( + self, message_id: str, payload: dict[str, Any], references: list[Any] + ) -> DeliveredEmail: + return DeliveredEmail( + message_id=message_id, + to=payload["to"], + cc=tuple(payload.get("cc", ())), + bcc=tuple(payload.get("bcc", ())), + subject=payload["subject"], + html=payload.get("html"), + text=payload.get("text"), + attachments=tuple( + DeliveredFile( + filename=reference["filename"], + content=self._bucket[reference["sha256"]], + content_type=reference.get("content_type"), + content_id=reference.get("content_id"), + ) + for reference in references + ), + ) + + +def lowercase(headers: Any) -> dict[str, str]: + return {name.lower(): value for name, value in headers.items()} + + +class FakePostHogAdapter(requests.adapters.BaseAdapter): + def __init__(self, posthog: FakePostHog) -> None: + super().__init__() + self.posthog = posthog + + def send( + self, request: requests.PreparedRequest, **kwargs: Any + ) -> requests.Response: + incoming = Incoming( + request.method or "", + request.url or "", + lowercase(request.headers), + request.body if isinstance(request.body, bytes) else b"", + ) + try: + answer = self.posthog.handle(incoming) + except Unreachable as exc: + raise requests.exceptions.ConnectionError(str(exc)) + response = requests.Response() + response.status_code = answer.status + response._content = answer.body + response.headers.update(answer.headers) + response.request = request + response.url = incoming.url + return response + + def close(self) -> None: + pass + + +def httpx_transport(posthog: FakePostHog) -> httpx.MockTransport: + def handle(request: httpx.Request) -> httpx.Response: + incoming = Incoming( + request.method, str(request.url), lowercase(request.headers), request.read() + ) + try: + answer = posthog.handle(incoming) + except Unreachable as exc: + raise httpx.ConnectError(str(exc), request=request) + return httpx.Response( + answer.status, content=answer.body, headers=answer.headers + ) + + return httpx.MockTransport(handle) + + +def session_on(posthog: FakePostHog, prefix: str = "https://") -> requests.Session: + session = requests.Session() + session.mount(prefix, FakePostHogAdapter(posthog)) + return session + + +def emails_on(posthog: FakePostHog, **options: Any) -> Emails: + return Emails(SECRET_KEY, session=session_on(posthog), **options) + + +def async_emails_on( + posthog: FakePostHog, max_retries: int = 2, **client_options: Any +) -> AsyncEmails: + return AsyncEmails( + SECRET_KEY, + max_retries=max_retries, + http_client=httpx.AsyncClient( + transport=httpx_transport(posthog), **client_options + ), + ) + + +INVITE_SHA256 = "215efae54e34ac5657da47b49c8d1bd8ea12649ee1efa838c9a47d7233f18afb" +LOGO = b"\x89PNG\r\n\x1a\n" +INVITE_FILE = {"filename": "invite.ics", "content": INVITE} +INVITE_AND_LOGO = [ + { + "filename": "invite.ics", + "content": INVITE, + "content_type": "text/calendar; method=REQUEST", + }, + {"filename": "logo.png", "content": "iVBORw0KGgo=", "content_id": "logo"}, +] +DELIVERED_INVITE_AND_LOGO = ( + DeliveredFile("invite.ics", INVITE, content_type="text/calendar; method=REQUEST"), + DeliveredFile("logo.png", LOGO, content_id="logo"), +) + + +def test_delivers_each_file_with_its_name_type_and_content_id(): + posthog = FakePostHog() + + result = send_code(emails_on(posthog), html=HTML, attachments=INVITE_AND_LOGO) + + [email] = posthog.delivered + assert result.id == email.message_id + assert email.attachments == DELIVERED_INVITE_AND_LOGO + + +def test_delivers_cc_and_bcc_beside_the_files(): + posthog = FakePostHog() + + send_code( + emails_on(posthog), + html=HTML, + cc="manager@example.com", + bcc=["audit@example.com", "archive@example.com"], + attachments=[INVITE_FILE], + ) + + [email] = posthog.delivered + assert (email.cc, email.bcc) == ( + ("manager@example.com",), + ("audit@example.com", "archive@example.com"), + ) + assert email.attachments == (DeliveredFile("invite.ics", INVITE),) + + +@parameterized.expand( + [ + ( + "one_address_each", + {"cc": "manager@example.com", "bcc": "audit@example.com"}, + ("manager@example.com",), + ("audit@example.com",), + ), + ( + "lists", + { + "cc": ["manager@example.com", "finance@example.com"], + "bcc": ["audit@example.com"], + }, + ("manager@example.com", "finance@example.com"), + ("audit@example.com",), + ), + ("empty_lists", {"cc": [], "bcc": [], "attachments": []}, (), ()), + ("none", {"cc": None, "bcc": None, "attachments": None}, (), ()), + ] +) +def test_delivers_cc_and_bcc_as_given(_name, overrides, cc, bcc): + posthog = FakePostHog() + + send_code(emails_on(posthog), html=HTML, **overrides) + + [email] = posthog.delivered + assert (email.to, email.cc, email.bcc, email.attachments) == ( + "user@example.com", + cc, + bcc, + (), + ) + + +@pytest.mark.asyncio +async def test_async_delivers_cc_and_bcc(): + posthog = FakePostHog() + + await send_code( + async_emails_on(posthog), + html=HTML, + cc="manager@example.com", + bcc=["audit@example.com", "archive@example.com"], + ) + + [email] = posthog.delivered + assert (email.cc, email.bcc) == ( + ("manager@example.com",), + ("audit@example.com", "archive@example.com"), + ) + + +def test_stores_only_the_files_posthog_lacks_under_their_own_hash(): + posthog = FakePostHog() + posthog.already_has(LOGO) + + send_code(emails_on(posthog), html=HTML, attachments=INVITE_AND_LOGO) + + assert posthog.stored_files == sorted([INVITE, LOGO]) + assert posthog.writes == 1 + [email] = posthog.delivered + assert email.attachments == DELIVERED_INVITE_AND_LOGO + + +def test_sends_without_uploading_when_posthog_has_every_file(): + posthog = FakePostHog() + posthog.already_has(INVITE) + + send_code(emails_on(posthog), html=HTML, attachments=[INVITE_FILE]) + + assert posthog.writes == 0 + [email] = posthog.delivered + assert email.attachments == (DeliveredFile("invite.ics", INVITE),) + + +def test_the_upload_carries_only_the_signed_url_and_headers_whatever_the_session_holds(): + posthog = FakePostHog() + session = session_on(posthog) + session.params = {"api_key": "session-default"} + session.headers.update( + {"Authorization": "Bearer session-default", "Content-Type": "text/plain"} + ) + + send_code(Emails(SECRET_KEY, session=session), html=HTML, attachments=[INVITE_FILE]) + + assert (posthog.stored_files, posthog.stored_content_types) == ([INVITE], [None]) + [email] = posthog.delivered + assert email.attachments == (DeliveredFile("invite.ics", INVITE),) + + +@parameterized.expand( + [ + ("bytearray", bytearray(INVITE)), + ("memoryview", memoryview(INVITE)), + ( + "strided_memoryview", + memoryview(bytes(b for byte in INVITE for b in (byte, 0)))[::2], + ), + ("base64_string", INVITE_BASE64), + ] +) +def test_file_content_from_bytes_or_base64_uploads_the_same_bytes(_name, content): + posthog = FakePostHog() + + send_code( + emails_on(posthog), + html=HTML, + attachments=[{"filename": "invite.ics", "content": content}], + ) + + assert posthog.stored_files == [INVITE] + [email] = posthog.delivered + assert email.attachments == (DeliveredFile("invite.ics", INVITE),) + + +def bytes_altered_on_the_way(posthog: FakePostHog) -> None: + posthog.inject("storage", alters_the_bytes) + + +def url_signed_16_minutes_ago(posthog: FakePostHog) -> None: + posthog.sign_urls_in_the_past(16) + + +@parameterized.expand( + [ + ( + "bytes_altered_on_the_way", + bytes_altered_on_the_way, + 400, + "BadDigest", + "The SHA256 you specified did not match the calculated checksum.", + ), + ( + "expired_url", + url_signed_16_minutes_ago, + 403, + "AccessDenied", + "Request has expired", + ), + ] +) +def test_a_refused_upload_sends_no_email_and_surfaces_the_storage_answer( + _name, arrange, status, storage_code, storage_message +): + posthog = FakePostHog() + arrange(posthog) + + with pytest.raises(EmailError) as raised: + send_code(emails_on(posthog), html=HTML, attachments=[INVITE_FILE]) + + error = raised.value + assert (error.name, error.code, error.status_code, error.field, error.why) == ( + "application_error", + "attachment_upload_failed", + status, + "attachments__0", + storage_message, + ) + assert storage_code in error.message + assert "proxy" in error.fix + assert (posthog.delivered, posthog.stored_files) == ([], []) + + +@parameterized.expand( + [ + ("line_wrapped", base64.encodebytes(bytes(100)).decode()), + ("unpadded", "QUI"), + ("url_safe", base64.urlsafe_b64encode(b"\xfb\xff").decode()), + ] +) +def test_refuses_base64_not_in_one_padded_line_and_says_how_to_encode(_name, content): + posthog = FakePostHog() + + with pytest.raises(EmailError) as raised: + send_code( + emails_on(posthog), + html=HTML, + attachments=[{"filename": "a.bin", "content": content}], + ) + + error = raised.value + assert (error.code, error.field) == ( + "invalid_attachment", + "attachments__0__content", + ) + assert "base64.b64encode" in error.message + assert (posthog.delivered, posthog.stored_files) == ([], []) + + +def self_signed_certificate(directory: Path) -> tuple[Path, Path]: + pytest.importorskip("cryptography") + from cryptography import x509 + from cryptography.hazmat.primitives import hashes, serialization + from cryptography.hazmat.primitives.asymmetric import ec + from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID + + key = ec.generate_private_key(ec.SECP256R1()) + name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "127.0.0.1")]) + now = datetime.now(timezone.utc) + certificate = ( + x509.CertificateBuilder() + .subject_name(name) + .issuer_name(name) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - timedelta(minutes=5)) + .not_valid_after(now + timedelta(days=1)) + .add_extension( + x509.SubjectAlternativeName([x509.IPAddress(ip_address("127.0.0.1"))]), + critical=False, + ) + .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True) + .add_extension( + x509.KeyUsage( + digital_signature=True, + content_commitment=False, + key_encipherment=False, + data_encipherment=False, + key_agreement=False, + key_cert_sign=True, + crl_sign=False, + encipher_only=False, + decipher_only=False, + ), + critical=True, + ) + .add_extension( + x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), critical=False + ) + .add_extension( + x509.SubjectKeyIdentifier.from_public_key(key.public_key()), critical=False + ) + .add_extension( + x509.AuthorityKeyIdentifier.from_issuer_public_key(key.public_key()), + critical=False, + ) + .sign(key, hashes.SHA256()) + ) + certificate_path, key_path = directory / "storage.pem", directory / "storage.key" + certificate_path.write_bytes(certificate.public_bytes(serialization.Encoding.PEM)) + key_path.write_bytes( + key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ) + ) + return certificate_path, key_path + + +def storage_handler(posthog: FakePostHog) -> type[BaseHTTPRequestHandler]: + class Storage(BaseHTTPRequestHandler): + def do_PUT(self) -> None: + body = self.rfile.read(int(self.headers.get("Content-Length") or 0)) + origin = f"https://127.0.0.1:{self.server.server_address[1]}" + answer = posthog.handle( + Incoming("PUT", origin + self.path, lowercase(self.headers), body) + ) + self.send_response(answer.status) + for name, value in answer.headers.items(): + self.send_header(name, value) + self.send_header("Content-Length", str(len(answer.body))) + self.end_headers() + self.wfile.write(answer.body) + + def log_message(self, *args: Any) -> None: + pass + + return Storage + + +@pytest.fixture +def posthog_with_tls_storage(tmp_path): + certificate, key = self_signed_certificate(tmp_path) + posthog = FakePostHog() + server = ThreadingHTTPServer(("127.0.0.1", 0), storage_handler(posthog)) + context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + context.minimum_version = ssl.TLSVersion.TLSv1_2 + context.load_cert_chain(certificate, key) + server.socket = context.wrap_socket(server.socket, server_side=True) + posthog.storage_url = f"https://127.0.0.1:{server.server_address[1]}/bucket" + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + yield posthog, certificate + finally: + server.shutdown() + server.server_close() + + +def test_the_upload_trusts_the_certificates_the_environment_names( + posthog_with_tls_storage, monkeypatch +): + posthog, certificate = posthog_with_tls_storage + monkeypatch.setenv("REQUESTS_CA_BUNDLE", str(certificate)) + emails = Emails(SECRET_KEY, session=session_on(posthog, POSTHOG_HOST)) + + send_code(emails, html=HTML, attachments=[INVITE_FILE]) + + assert posthog.stored_files == [INVITE] + [email] = posthog.delivered + assert email.attachments == (DeliveredFile("invite.ics", INVITE),) + + +def test_the_upload_refuses_a_storage_whose_certificate_nothing_trusts( + posthog_with_tls_storage, monkeypatch +): + posthog, _certificate = posthog_with_tls_storage + monkeypatch.delenv("REQUESTS_CA_BUNDLE", raising=False) + monkeypatch.delenv("CURL_CA_BUNDLE", raising=False) + emails = Emails(SECRET_KEY, session=session_on(posthog, POSTHOG_HOST)) + + with pytest.raises(EmailError) as raised: + send_code(emails, html=HTML, attachments=[INVITE_FILE]) + + assert (raised.value.code, raised.value.field) == ( + "network_error", + "attachments__0", + ) + assert (posthog.delivered, posthog.stored_files) == ([], []) + + +def test_an_unreachable_storage_never_shows_the_signed_url_in_the_traceback(): + posthog = FakePostHog() + posthog.inject("storage", unreachable, unreachable, unreachable) + + with pytest.raises(EmailError) as raised: + send_code(emails_on(posthog), html=HTML, attachments=[INVITE_FILE]) + + error = raised.value + assert (error.code, error.field) == ("network_error", "attachments__0") + assert "file storage" in error.message + assert "X-Amz-Signature" not in "".join(traceback.format_exception(error)) + assert (posthog.delivered, posthog.stored_files) == ([], []) + + +@pytest.mark.asyncio +async def test_async_unreachable_storage_never_shows_the_signed_url_in_the_traceback(): + posthog = FakePostHog() + posthog.inject("storage", unreachable) + + with pytest.raises(EmailError) as raised: + await send_code( + async_emails_on(posthog, max_retries=0), + html=HTML, + attachments=[INVITE_FILE], + ) + + error = raised.value + assert (error.code, error.field) == ("network_error", "attachments__0") + assert "file storage" in error.message + assert "X-Amz-Signature" not in "".join(traceback.format_exception(error)) + assert (posthog.delivered, posthog.stored_files) == ([], []) + + +@parameterized.expand( + [ + ( + "slow_down", + answers(storage_error(503, "SlowDown", "Please reduce your request rate.")), + ), + ( + "internal_error", + answers(storage_error(500, "InternalError", "Please try again.")), + ), + ( + "too_many_requests", + answers(storage_error(429, "SlowDown", "Please reduce your request rate.")), + ), + ("connection_dropped", unreachable), + ] +) +def test_a_storage_that_asks_for_a_retry_is_retried_before_the_send(_name, fault): + posthog = FakePostHog() + posthog.inject("storage", fault) + + send_code(emails_on(posthog), html=HTML, attachments=[INVITE_FILE]) + + assert posthog.stored_files == [INVITE] + [email] = posthog.delivered + assert email.attachments == (DeliveredFile("invite.ics", INVITE),) + + +def test_a_project_that_cannot_send_files_uploads_and_sends_nothing(): + posthog = FakePostHog() + posthog.project_can_send_files = False + + with pytest.raises(EmailError) as raised: + send_code(emails_on(posthog), html=HTML, attachments=[INVITE_FILE]) + + error = raised.value + assert (error.code, error.field, error.why) == ( + "field_not_accepted", + "attachments", + "Sending files isn't available to this project yet.", + ) + assert (posthog.delivered, posthog.stored_files) == ([], []) + + +def test_an_upload_refusal_naming_a_file_names_the_first_attachment_with_those_bytes(): + posthog = FakePostHog() + posthog.inject( + "uploads", + answers( + api_error( + 400, + "message_too_large", + "This file is too large to send.", + attr="files__1__size", + ) + ), + ) + + with pytest.raises(EmailError) as raised: + send_code( + emails_on(posthog), + html=HTML, + attachments=[ + INVITE_FILE, + {"filename": "invite-copy.ics", "content": INVITE}, + {"filename": "logo.png", "content": LOGO}, + ], + ) + + assert (raised.value.code, raised.value.field) == ( + "message_too_large", + "attachments__2", + ) + assert (posthog.delivered, posthog.stored_files) == ([], []) + + +@parameterized.expand( + [ + ("throttled", 429, "SlowDown", "rate_limit_exceeded"), + ("unavailable", 503, "ServiceUnavailable", "service_unavailable"), + ] +) +def test_a_storage_out_of_retries_names_its_passing_fault( + _name, status, storage_code, name +): + posthog = FakePostHog() + posthog.inject( + "storage", answers(storage_error(status, storage_code, "Try again.")) + ) + + with pytest.raises(EmailError) as raised: + send_code( + emails_on(posthog, max_retries=0), html=HTML, attachments=[INVITE_FILE] + ) + + assert (raised.value.name, raised.value.code) == (name, "attachment_upload_failed") + assert (posthog.delivered, posthog.stored_files) == ([], []) + + +@parameterized.expand( + [ + ("posthog_without_the_upload_route", "Endpoint not found."), + ("sending_from_code_off_for_the_project", "Not found."), + ] +) +def test_an_upload_route_404_reports_posthogs_answer_on_attachments_and_sends_nothing( + _name, detail +): + posthog = FakePostHog() + posthog.inject( + "uploads", + answers(api_error(404, "not_found", detail, kind="invalid_request", attr=None)), + ) + + with pytest.raises(EmailError) as raised: + send_code(emails_on(posthog), html=HTML, attachments=[INVITE_FILE]) + + error = raised.value + assert (error.name, error.code, error.status_code, error.field, error.message) == ( + "not_found", + "not_found", + 404, + "attachments", + detail, + ) + assert error.why and error.fix + assert (posthog.delivered, posthog.stored_files) == ([], []) + + +def test_an_upload_route_404_without_an_answer_still_names_attachments(): + posthog = FakePostHog() + posthog.inject("uploads", answers(Answer(404, b"

Not Found

"))) + + with pytest.raises(EmailError) as raised: + send_code(emails_on(posthog), html=HTML, attachments=[INVITE_FILE]) + + error = raised.value + assert (error.code, error.status_code, error.field) == ( + "not_found", + 404, + "attachments", + ) + assert (posthog.delivered, posthog.stored_files) == ([], []) + + +def test_a_project_over_its_daily_upload_bytes_is_told_without_retrying(): + posthog = FakePostHog() + posthog.inject( + "uploads", + answers( + api_error( + 429, + "upload_quota_exceeded", + "The project uploaded its daily limit of file bytes.", + kind="throttled_error", + ) + ), + ) + + with pytest.raises(EmailError) as raised: + send_code(emails_on(posthog), html=HTML, attachments=[INVITE_FILE]) + + assert (raised.value.code, raised.value.name) == ( + "upload_quota_exceeded", + "daily_quota_exceeded", + ) + assert (posthog.delivered, posthog.stored_files) == ([], []) + + +def test_a_send_retried_after_a_lost_reply_replays_with_its_idempotency_key(): + posthog = FakePostHog() + posthog.inject( + "send", + loses_the_reply( + api_error(503, "delivery_unavailable", "Try again.", kind="server_error") + ), + ) + + result = send_code( + emails_on(posthog), + html=HTML, + idempotency_key="invite:user_123", + attachments=[INVITE_FILE], + ) + + [email] = posthog.delivered + assert (result.id, result.replayed) == (email.message_id, True) + assert email.attachments == (DeliveredFile("invite.ics", INVITE),) + + +def test_the_same_bytes_in_two_attachments_upload_once(): + posthog = FakePostHog() + + send_code( + emails_on(posthog), + html=HTML, + attachments=[ + {"filename": "logo.png", "content": LOGO, "content_id": "logo"}, + {"filename": "logo-copy.png", "content": LOGO}, + ], + ) + + assert (posthog.stored_files, posthog.writes) == ([LOGO], 1) + [email] = posthog.delivered + assert email.attachments == ( + DeliveredFile("logo.png", LOGO, content_id="logo"), + DeliveredFile("logo-copy.png", LOGO), + ) + + +def test_a_failed_upload_names_the_first_attachment_with_those_bytes(): + posthog = FakePostHog() + posthog.already_has(INVITE) + posthog.sign_urls_in_the_past(16) + + with pytest.raises(EmailError) as raised: + send_code( + emails_on(posthog), + html=HTML, + attachments=[ + INVITE_FILE, + {"filename": "logo.png", "content": LOGO}, + {"filename": "logo-copy.png", "content": LOGO}, + ], + ) + + assert (raised.value.code, raised.value.field) == ( + "attachment_upload_failed", + "attachments__1", + ) + assert posthog.delivered == [] + + +def test_refuses_eleven_files_before_reading_any_of_them(): + posthog = FakePostHog() + attachments: list[Any] = [ + {"filename": f"{i}.txt", "content": b"hi"} for i in range(10) + ] + + with pytest.raises(EmailError) as raised: + send_code( + emails_on(posthog), + html=HTML, + attachments=attachments + [{"filename": "late.txt", "content": None}], + ) + + assert raised.value.code == "too_many_attachments" + assert (posthog.delivered, posthog.stored_files) == ([], []) + + +@parameterized.expand( + [ + ( + "eleven_files", + [{"filename": f"{i}.txt", "content": b"hi"} for i in range(11)], + "too_many_attachments", + ), + ( + "one_file_too_large_beside_the_body", + [{"filename": "scan.pdf", "content": bytes(29_182_821)}], + "message_too_large", + ), + ( + "files_too_large_together", + [{"filename": f"{i}.pdf", "content": bytes(3_000_000)} for i in range(10)], + "message_too_large", + ), + ] +) +def test_refuses_an_email_over_the_limits_before_any_request(_name, attachments, code): + posthog = FakePostHog() + + with pytest.raises(EmailError) as raised: + send_code(emails_on(posthog), html="Hi!", attachments=attachments) + + error = raised.value + assert (error.name, error.code, error.field) == ( + "validation_error", + code, + "attachments", + ) + assert error.fix + assert (posthog.delivered, posthog.stored_files) == ([], []) + + +def test_refuses_an_oversized_email_without_copying_its_files(): + posthog = FakePostHog() + emails = emails_on(posthog) + scan = memoryview(bytearray(8 * 1024 * 1024)) + + tracemalloc.start() + try: + with pytest.raises(EmailError) as raised: + send_code( + emails, + html=HTML, + attachments=[ + {"filename": f"scan-{i}.pdf", "content": scan} for i in range(10) + ], + ) + _current, peak = tracemalloc.get_traced_memory() + finally: + tracemalloc.stop() + + assert raised.value.code == "message_too_large" + assert peak < len(scan) + assert posthog.stored_files == [] + assert posthog.delivered == [] + + +def test_refuses_oversized_padded_base64_without_copying_it(): + posthog = FakePostHog() + emails = emails_on(posthog) + scan = "AAAA" * 10_000_000 + "AA==" + + tracemalloc.start() + try: + with pytest.raises(EmailError) as raised: + send_code( + emails, + html=HTML, + attachments=[{"filename": "scan.pdf", "content": scan}], + ) + _current, peak = tracemalloc.get_traced_memory() + finally: + tracemalloc.stop() + + assert raised.value.code == "message_too_large" + assert peak < len(scan) // 4 + assert posthog.stored_files == [] + assert posthog.delivered == [] + + +def test_refuses_extra_base64_padding_without_copying_it(): + posthog = FakePostHog() + emails = emails_on(posthog) + content = "YWJj" + "=" * 40_000_000 + + tracemalloc.start() + try: + with pytest.raises(EmailError) as raised: + send_code( + emails, + html=HTML, + attachments=[{"filename": "a.txt", "content": content}], + ) + _current, peak = tracemalloc.get_traced_memory() + finally: + tracemalloc.stop() + + assert (raised.value.code, raised.value.field) == ( + "invalid_attachment", + "attachments__0__content", + ) + assert "base64.b64encode" in raised.value.message + assert peak < len(content) // 4 + assert posthog.stored_files == [] + assert posthog.delivered == [] + + +LARGEST_SCAN = bytes(29_182_815) + + +def test_sends_the_largest_file_that_fits_beside_the_body(): + posthog = FakePostHog() + + send_code( + emails_on(posthog), + html="Hi!", + attachments=[{"filename": "scan.pdf", "content": LARGEST_SCAN}], + ) + + [email] = posthog.delivered + assert email.attachments == (DeliveredFile("scan.pdf", LARGEST_SCAN),) + + +SCAN_BESIDE_A_78_BYTE_BODY = bytes(29_182_761) + + +@parameterized.expand( + [ + ("html", "html", "text"), + ("text", "text", "html"), + ] +) +def test_counts_a_body_like_a_file_in_the_size_estimate( + _name, body_field, other_body_field +): + posthog = FakePostHog() + emails = emails_on(posthog) + attachments = [{"filename": "scan.pdf", "content": SCAN_BESIDE_A_78_BYTE_BODY}] + + send_code( + emails, attachments=attachments, **{body_field: "a" * 57, other_body_field: "b"} + ) + with pytest.raises(EmailError) as raised: + send_code( + emails, + attachments=attachments, + **{body_field: "é" * 57, other_body_field: "b"}, + ) + + assert raised.value.code == "message_too_large" + assert len(posthog.delivered) == 1 + + +def test_counts_the_html_twice_without_text_since_posthog_derives_the_text(): + posthog = FakePostHog() + emails = emails_on(posthog) + attachments = [{"filename": "scan.pdf", "content": SCAN_BESIDE_A_78_BYTE_BODY}] + + with pytest.raises(EmailError) as raised: + send_code(emails, html="a" * 57, attachments=attachments) + send_code(emails, html="a" * 57, text="b", attachments=attachments) + + assert raised.value.code == "message_too_large" + assert len(posthog.delivered) == 1 + + +def test_uploads_only_to_an_https_storage_url(): + posthog = FakePostHog(storage_url="http://files.example.com/team/1") + + with pytest.raises(EmailError) as raised: + send_code( + emails_on(posthog), + html=HTML, + attachments=[ + {"filename": "notes.txt", "content": b"notes"}, + {"filename": "invite.ics", "content": INVITE}, + ], + ) + + error = raised.value + assert (error.code, error.field) == ("api_error", "attachments__0") + assert "https" in error.message and "file storage" in (error.fix or "") + assert "files.example.com" not in str(error) + str(vars(error)) + assert (posthog.delivered, posthog.stored_files, posthog.writes) == ([], [], 0) + + +def test_a_plain_http_host_for_local_development_uploads_over_plain_http(): + posthog = FakePostHog( + storage_url="http://localhost:8333/team/1", + posthog_host="http://localhost:8000", + ) + emails = Emails( + SECRET_KEY, + host="http://localhost:8000", + session=session_on(posthog, "http://"), + ) + + send_code( + emails, html=HTML, attachments=[{"filename": "invite.ics", "content": INVITE}] + ) + + [email] = posthog.delivered + assert email.attachments == (DeliveredFile("invite.ics", INVITE),) + + +@parameterized.expand( + [ + ("42_ascii_bytes_take_75", "a" * 42, 29_182_815), + ("43_ascii_bytes_take_150", "a" * 43, 29_182_758), + ("21_two_byte_characters_take_75", "é" * 21, 29_182_815), + ("22_two_byte_characters_take_150", "é" * 22, 29_182_758), + ] +) +def test_a_subject_takes_75_bytes_of_the_limit_per_42_utf8_bytes( + _name, subject, largest_file_size +): + posthog = FakePostHog() + emails = emails_on(posthog) + + send_code( + emails, + html="Hi!", + subject=subject, + attachments=[{"filename": "scan.pdf", "content": bytes(largest_file_size)}], + ) + with pytest.raises(EmailError) as raised: + send_code( + emails, + html="Hi!", + subject=subject, + attachments=[ + {"filename": "scan.pdf", "content": bytes(largest_file_size + 1)} + ], + ) + + assert raised.value.code == "message_too_large" + assert len(posthog.delivered) == 1 + + +class PostHogWithARedirectingStorage(BaseHTTPRequestHandler): + emails_sent = 0 + redirect_followed = False + + def do_POST(self) -> None: + self.rfile.read(int(self.headers["Content-Length"])) + if self.path.endswith("/attachments/uploads/"): + slot = { + "sha256": INVITE_SHA256, + "status": "upload", + "url": f"http://{self.headers['Host']}/bucket/{INVITE_SHA256}", + "method": "PUT", + "headers": {"content-length": str(len(INVITE))}, + "url_expires_at": "2026-11-09T12:15:00Z", + } + self.answer(200, json.dumps({"files": [slot]}).encode()) + else: + type(self).emails_sent += 1 + self.answer(202, json.dumps({"message_id": "msg_1"}).encode()) + + def do_PUT(self) -> None: + self.rfile.read(int(self.headers["Content-Length"])) + self.send_response(302) + self.send_header("Location", "/login") + self.send_header("Content-Length", "0") + self.end_headers() + + def do_GET(self) -> None: + type(self).redirect_followed = True + self.answer(200, b"Sign in") + + def answer(self, status: int, body: bytes) -> None: + self.send_response(status) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args: Any) -> None: + pass + + +class PostHogWithAStorageThatHangsUp(PostHogWithARedirectingStorage): + def do_PUT(self) -> None: + self.rfile.read(int(self.headers["Content-Length"])) + self.send_response(500) + self.send_header("Content-Length", "100") + self.end_headers() + self.wfile.write(b"") + self.close_connection = True + + +def test_a_storage_answer_cut_short_names_the_attachment_even_on_a_streaming_session(): + server = ThreadingHTTPServer(("127.0.0.1", 0), PostHogWithAStorageThatHangsUp) + PostHogWithAStorageThatHangsUp.emails_sent = 0 + threading.Thread(target=server.serve_forever, daemon=True).start() + session = requests.Session() + session.stream = True + session.trust_env = False + emails = Emails( + SECRET_KEY, + host=f"http://127.0.0.1:{server.server_address[1]}", + max_retries=0, + session=session, + ) + + try: + with pytest.raises(EmailError) as raised: + send_code(emails, html=HTML, attachments=[INVITE_FILE]) + finally: + server.shutdown() + server.server_close() + + assert (raised.value.code, raised.value.field) == ( + "network_error", + "attachments__0", + ) + assert PostHogWithAStorageThatHangsUp.emails_sent == 0 + + +@pytest.mark.asyncio +async def test_async_upload_never_follows_a_storage_redirect_even_when_the_client_does(): + server = ThreadingHTTPServer(("127.0.0.1", 0), PostHogWithARedirectingStorage) + PostHogWithARedirectingStorage.emails_sent = 0 + PostHogWithARedirectingStorage.redirect_followed = False + threading.Thread(target=server.serve_forever, daemon=True).start() + emails = AsyncEmails( + SECRET_KEY, + host=f"http://127.0.0.1:{server.server_address[1]}", + max_retries=0, + http_client=httpx.AsyncClient(follow_redirects=True, trust_env=False), + ) + + try: + with pytest.raises(EmailError) as raised: + await emails.send( + to="user@example.com", + template_key="password-reset", + subject="Reset your password", + html=HTML, + attachments=[{"filename": "invite.ics", "content": INVITE}], + ) + finally: + server.shutdown() + server.server_close() + + assert (raised.value.code, raised.value.field) == ( + "attachment_upload_failed", + "attachments__0", + ) + assert PostHogWithARedirectingStorage.emails_sent == 0 + assert not PostHogWithARedirectingStorage.redirect_followed + + +class ServerElsewhere(BaseHTTPRequestHandler): + requests_received = 0 + + def do_PUT(self) -> None: + type(self).requests_received += 1 + self.rfile.read(int(self.headers.get("Content-Length", 0))) + self.send_response(200) + self.send_header("Content-Length", "0") + self.end_headers() + + do_GET = do_POST = do_PUT + + def log_message(self, *args: Any) -> None: + pass + + +class PostHogWithStorageRedirectingElsewhere(PostHogWithARedirectingStorage): + elsewhere = "" + + def do_PUT(self) -> None: + self.rfile.read(int(self.headers["Content-Length"])) + self.send_response(307) + self.send_header("Location", f"{self.elsewhere}/bucket/copy") + self.send_header("Content-Length", "0") + self.end_headers() + + +@contextmanager +def serving(handler: type[BaseHTTPRequestHandler]) -> Iterator[str]: + server = ThreadingHTTPServer(("127.0.0.1", 0), handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + yield f"http://127.0.0.1:{server.server_address[1]}" + finally: + server.shutdown() + server.server_close() + + +@contextmanager +def posthog_redirecting_storage_elsewhere() -> Iterator[tuple[str, str]]: + ServerElsewhere.requests_received = 0 + PostHogWithStorageRedirectingElsewhere.emails_sent = 0 + with serving(ServerElsewhere) as elsewhere: + PostHogWithStorageRedirectingElsewhere.elsewhere = elsewhere + with serving(PostHogWithStorageRedirectingElsewhere) as posthog: + yield posthog, elsewhere + + +def error_text(error: EmailError) -> str: + return " ".join(str(part) for part in (error, error.message, error.why, error.fix)) + + +def test_an_upload_never_follows_storage_to_another_server(): + session = requests.Session() + session.trust_env = False + + with posthog_redirecting_storage_elsewhere() as (posthog, elsewhere): + with pytest.raises(EmailError) as raised: + send_code( + Emails(SECRET_KEY, host=posthog, max_retries=0, session=session), + html=HTML, + attachments=[INVITE_FILE], + ) + + assert (raised.value.code, raised.value.field) == ( + "attachment_upload_failed", + "attachments__0", + ) + assert elsewhere not in error_text(raised.value) + assert "/bucket/" not in error_text(raised.value) + assert ServerElsewhere.requests_received == 0 + assert PostHogWithStorageRedirectingElsewhere.emails_sent == 0 + + +@pytest.mark.asyncio +async def test_async_upload_never_follows_storage_to_another_server(): + with posthog_redirecting_storage_elsewhere() as (posthog, elsewhere): + emails = AsyncEmails( + SECRET_KEY, + host=posthog, + max_retries=0, + http_client=httpx.AsyncClient(follow_redirects=True, trust_env=False), + ) + with pytest.raises(EmailError) as raised: + await emails.send( + to="user@example.com", + template_key="password-reset", + subject="Reset your password", + html=HTML, + attachments=[INVITE_FILE], + ) + + assert (raised.value.code, raised.value.field) == ( + "attachment_upload_failed", + "attachments__0", + ) + assert elsewhere not in error_text(raised.value) + assert "/bucket/" not in error_text(raised.value) + assert ServerElsewhere.requests_received == 0 + assert PostHogWithStorageRedirectingElsewhere.emails_sent == 0 + + +@pytest.mark.asyncio +async def test_async_upload_carries_only_the_signed_url_and_headers_whatever_the_client_holds(): + posthog = FakePostHog() + emails = async_emails_on( + posthog, + params={"api_key": "client-default"}, + headers={ + "Authorization": "Bearer client-default", + "Content-Type": "text/plain", + }, + ) + + result = await send_code(emails, html=HTML, attachments=INVITE_AND_LOGO) + + assert (posthog.stored_files, posthog.stored_content_types) == ( + sorted([INVITE, LOGO]), + [None, None], + ) + [email] = posthog.delivered + assert result.id == email.message_id + assert email.attachments == DELIVERED_INVITE_AND_LOGO + + +@pytest.mark.asyncio +async def test_async_refused_upload_sends_no_email(): + posthog = FakePostHog() + posthog.sign_urls_in_the_past(16) + + with pytest.raises(EmailError) as raised: + await send_code(async_emails_on(posthog), html=HTML, attachments=[INVITE_FILE]) + + assert (raised.value.code, raised.value.field, raised.value.status_code) == ( + "attachment_upload_failed", + "attachments__0", + 403, + ) + assert (posthog.delivered, posthog.stored_files) == ([], []) + + +SECRET_CONTENT = b"secret-file-content" + + +@parameterized.expand( + [ + ( + "one_file_not_a_list", + {"filename": "a.txt", "content": SECRET_CONTENT}, + "attachments", + ), + ("bytes_not_a_list", SECRET_CONTENT, "attachments"), + ("file_not_a_mapping", [SECRET_CONTENT], "attachments__0"), + ( + "list_of_ints", + [{"filename": "a.txt", "content": list(SECRET_CONTENT)}], + "attachments__0__content", + ), + ("no_content", [{"filename": "a.txt"}], "attachments__0__content"), + ( + "empty_content", + [{"filename": "a.txt", "content": b""}], + "attachments__0__content", + ), + ( + "text_not_base64", + [{"filename": "a.txt", "content": "secret-file-content"}], + "attachments__0__content", + ), + ( + "non_ascii_text", + [{"filename": "a.txt", "content": "secret-file-content é"}], + "attachments__0__content", + ), + ("no_filename", [{"content": SECRET_CONTENT}], "attachments__0__filename"), + ( + "empty_filename", + [{"filename": "", "content": SECRET_CONTENT}], + "attachments__0__filename", + ), + ( + "path", + [{"filename": "a.txt", "content": SECRET_CONTENT, "path": "/tmp/a.txt"}], + "attachments__0__path", + ), + ( + "unknown_field", + [ + { + "filename": "a.txt", + "content": SECRET_CONTENT, + "contentType": "text/plain", + } + ], + "attachments__0__contentType", + ), + ( + "none_key_before_an_unknown_field", + [ + { + None: "ignored", + "filename": "a.txt", + "content": SECRET_CONTENT, + "contentType": "text/plain", + } + ], + "attachments__0__None", + ), + ( + "content_type_not_text", + [ + {"filename": "a.txt", "content": "aGk="}, + {"filename": "b.txt", "content": SECRET_CONTENT, "content_type": 1}, + ], + "attachments__1__content_type", + ), + ] +) +def test_rejects_an_attachment_it_cannot_send_without_repeating_its_content( + _name, attachments, field +): + posthog = FakePostHog() + + with pytest.raises(EmailError) as raised: + send_code(emails_on(posthog), html=HTML, attachments=attachments) + + error = raised.value + assert (error.name, error.code, error.field) == ( + "validation_error", + "invalid_attachment", + field, + ) + assert "secret-file-content" not in str(error) + assert "115, 101" not in str(error) + assert (posthog.delivered, posthog.stored_files) == ([], []) diff --git a/references/public_api_snapshot.txt b/references/public_api_snapshot.txt index 8eec881b9..5096b4f34 100644 --- a/references/public_api_snapshot.txt +++ b/references/public_api_snapshot.txt @@ -912,7 +912,10 @@ attribute posthog.debug = False attribute posthog.default_client = None attribute posthog.disable_geoip = True attribute posthog.disabled = False +attribute posthog.emails.CodeEmailArgs.attachments: NotRequired[Optional[Sequence[EmailAttachment]]] +attribute posthog.emails.CodeEmailArgs.bcc: NotRequired[Optional[Union[str, Sequence[str]]]] attribute posthog.emails.CodeEmailArgs.capture_engagement: NotRequired[Optional[bool]] +attribute posthog.emails.CodeEmailArgs.cc: NotRequired[Optional[Union[str, Sequence[str]]]] attribute posthog.emails.CodeEmailArgs.from_: NotRequired[Optional[str]] attribute posthog.emails.CodeEmailArgs.html: NotRequired[Optional[str]] attribute posthog.emails.CodeEmailArgs.reply_to: NotRequired[Optional[Union[str, Sequence[str]]]] @@ -921,6 +924,10 @@ attribute posthog.emails.CodeEmailArgs.subject: str attribute posthog.emails.CodeEmailArgs.template_key: str attribute posthog.emails.CodeEmailArgs.text: NotRequired[Optional[str]] attribute posthog.emails.CodeEmailArgs.tracking: NotRequired[Optional[Union[bool, EmailTracking]]] +attribute posthog.emails.EmailAttachment.content: Union[bytes, bytearray, memoryview, str] +attribute posthog.emails.EmailAttachment.content_id: NotRequired[str] +attribute posthog.emails.EmailAttachment.content_type: NotRequired[str] +attribute posthog.emails.EmailAttachment.filename: str attribute posthog.emails.EmailError.code = code attribute posthog.emails.EmailError.docs_url = docs_url attribute posthog.emails.EmailError.field = field @@ -1376,6 +1383,7 @@ class posthog.consumer.Consumer(queue, api_key, flush_at=100, host=None, on_erro class posthog.contexts.ContextScope(parent=None, fresh: bool = False, capture_exceptions: bool = True, client: Optional[Client] = None) class posthog.emails.AsyncEmails(secret_key: Optional[str], *, host: Optional[str] = None, timeout: float = 15, max_retries: int = 2, disabled: bool = False, http_client: Any = None, migrated_from: Optional[str] = None) class posthog.emails.CodeEmailArgs +class posthog.emails.EmailAttachment class posthog.emails.EmailError(*, name: str, code: str, message: str, status_code: Optional[int] = None, why: Optional[str] = None, fix: Optional[str] = None, docs_url: Optional[str] = None, field: Optional[str] = None, retry_after: Optional[float] = None) class posthog.emails.EmailStatus(id: str, status: str, created_at: datetime, status_at: datetime, reason: Optional[str] = None) class posthog.emails.EmailTracking diff --git a/typings/requests/__init__.pyi b/typings/requests/__init__.pyi index 1c545b680..91e583b5e 100644 --- a/typings/requests/__init__.pyi +++ b/typings/requests/__init__.pyi @@ -6,10 +6,22 @@ from .auth import AuthBase as _AuthBase class PreparedRequest: headers: dict[str, str] +class Request: + def __init__( + self, + method: str, + url: str, + *, + data: bytes | None = ..., + headers: dict[str, str] | None = ..., + ) -> None: ... + def prepare(self) -> PreparedRequest: ... + class Response: status_code: int ok: bool text: str + content: bytes headers: dict[str, str] def json(self) -> Any: ... def close(self) -> None: ... @@ -19,6 +31,22 @@ class Session: headers: dict[str, str] def mount(self, prefix: str, adapter: adapters.HTTPAdapter) -> None: ... def close(self) -> None: ... + def merge_environment_settings( + self, + url: str, + proxies: dict[str, str], + stream: bool | None, + verify: bool | str | None, + cert: str | None, + ) -> dict[str, Any]: ... + def send( + self, + request: PreparedRequest, + *, + timeout: float | None = ..., + allow_redirects: bool = ..., + **kwargs: Any, + ) -> Response: ... def request( self, method: str,