diff --git a/CHANGELOG.md b/CHANGELOG.md index 40151a3..7ce2cbb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,33 @@ Postmaster MCP follows Semantic Versioning for stable releases. Every stable release should update `VERSION`, this changelog, and publish an immutable Git tag/release named `vX.Y.Z`. +## 9.4.0 - 2026-08-19 + +### Added +- Per-link HTTP/HTTPS click tracking with random opaque per-delivery occurrence tokens and the public `GET /t/c/` redirect endpoint. Redirect destinations are resolved only from server-side records, so query parameters cannot turn the endpoint into a generic open redirect. +- Additive `tracking_links` and `tracking_clicks` analytics tables with campaign/delivery/message correlation, recipient, original and normalized URL, destination host, anchor index/label, UTC timestamps and the existing country/source/browser/OS/User-Agent/fingerprint enrichment pipeline. +- Stable unique-click definition: `delivery_id + link_id + client_fingerprint`, with the existing keyed fingerprint fallback when network/User-Agent inputs are unavailable. +- Link analytics for total/unique clicks, unique recipients, first/last click, top links, campaign/delivery/link filtering and unified pixel/AMP/link event detail. +- MCP read tools `get_tracking_summary`, `list_tracking_links` and `list_tracking_events`; existing `tracking_status` and `get_tracking_campaign` are extended without removing legacy tools. +- Tracking dashboard `Top links` and unified `Tracking events` views while preserving the existing pixel/open dashboard sections. +- Clean Sent-copy generation for individualized tracked/AMP deliveries. Recipient MIME keeps tracking instrumentation; archived Sent MIME keeps original links and omits recipient pixel/click/AMP callback instrumentation. +- `build_status.link_tracking` and `build_status.sent_copy_tracking_sanitized` capability flags. +- `docs/LINK_TRACKING.md` covering architecture, schema, unique clicks, Sent-clean behavior, Cloudflare Access and live deployment preflight. + +### Changed +- Link instrumentation is applied to the same existing tracking opt-in used by `track_opens`, preserving current per-send/account-default privacy semantics rather than adding another required send parameter. +- Tracked recipient and Sent MIME variants are built independently from the same canonical body/attachment inputs. `Message-ID`/`Date` are synchronized and normal threading headers are preserved; serialized MIME is not sanitized with fragile regex replacement. +- The v9.3 single-YAML Portainer bootstrap remains unchanged. With `POSTMASTER_VERSION=latest` and `POSTMASTER_CHECK_UPDATES_ON_START=true`, a stack restart is sufficient to select v9.4.0 after the stable release is published. + +### Fixed +- Viewing a newly generated tracked message in the sender's Sent mailbox no longer loads the recipient tracking pixel. +- Clicking a link in a newly generated Sent copy no longer traverses the recipient `/t/c/` URL, preventing sender self-clicks from being attributed to the recipient. + +### Security / deployment +- Existing public pixel and AMP callback paths remain unchanged. v9.4 requires exactly one new anonymous Cloudflare Access bypass: `/t/c/*`. +- `/mcp`, dashboard/admin/private APIs, mail/task/memory/skill/file-management routes and tracking analytics remain protected. +- `/files/*` is a pre-existing v9.3 signed file-handoff concern and is not added to the v9.4 Cloudflare bypass policy. + ## 9.3.0 - 2026-08-18 ### Added diff --git a/README.md b/README.md index 481bb5c..d9e010e 100644 --- a/README.md +++ b/README.md @@ -653,3 +653,35 @@ The existing `PUBLIC_MCP_HOST` is reused as the normal HTTPS base for the same s An existing stack with `POSTMASTER_VERSION=latest` can therefore receive v9.3 by restarting after the stable release is published. If the external access layer protects the full app, ensure the signed `/files/*` route is reachable according to the deployment's proxy policy without weakening protection for `/mcp`, the dashboard, or unrelated routes. See `docs/FILE_HANDOFF.md` for the handoff hierarchy, security model, signed URL behavior and deployment details. + +# Per-link tracking and clean Sent copies (v9.4) + +v9.4 adds per-link HTTP/HTTPS click telemetry to the existing tracked-delivery pipeline while leaving the existing `/track/open/*` pixel behavior unchanged. Eligible recipient HTML anchors are rewritten to opaque URLs under: + +```text +/t/c/ +``` + +The random token identifies a server-side link occurrence; recipient data and destination URLs are not encoded into it. The click endpoint resolves the stored record, records a `link` event with the existing country/source/browser/OS/User-Agent/client-fingerprint enrichment and immediately redirects to the exact stored `original_url`. Query parameters supplied to `/t/c/` are never accepted as redirect destinations. + +The v9.4 unique-click definition is: + +```text +delivery_id + link_id + client_fingerprint +``` + +Analytics expose total clicks, unique clicks, unique recipients, first/last click, destination host, per-campaign/per-delivery/per-link filtering and top links. Existing `tracking_status` and `get_tracking_campaign` are extended, while `get_tracking_summary`, `list_tracking_links` and `list_tracking_events` provide read-only link/event queries. Opaque tracking tokens are not returned by list/dashboard APIs. + +Recipient and Sent MIME are now generated separately from the same canonical body and attachment inputs. The recipient copy keeps the existing tracking pixel and tracked URLs; the archived Sent copy keeps original URLs and contains no active recipient pixel or `/t/c/`. `Message-ID`, Date and normal threading headers are preserved and attachments reuse the same original bytes. This prevents sender self-opens/self-clicks from being attributed to the recipient for messages generated by v9.4+. + +The single-YAML bootstrap is unchanged. With `POSTMASTER_VERSION=latest` and `POSTMASTER_CHECK_UPDATES_ON_START=true`, restart the stack after the stable release is published. + +Cloudflare Access is external to the container. Keep the existing public bypasses for `/track/open/*` and `/api/amp/*`, and add exactly one new public bypass required by v9.4: + +```text +/t/c/* +``` + +Do not expose `/mcp`, dashboard/admin/private APIs, mail/task/memory/skill/file-management endpoints or tracking analytics. The pre-existing v9.3 signed `/files/*` handoff remains a separate deployment-policy concern and is not added automatically as part of v9.4. + +See `docs/LINK_TRACKING.md` for architecture, schema, Sent-clean behavior, analytics and the live Cloudflare preflight. diff --git a/VERSION b/VERSION index b13d146..8148c55 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -9.3.0 +9.4.0 diff --git a/docs/LINK_TRACKING.md b/docs/LINK_TRACKING.md new file mode 100644 index 0000000..8ac5d3e --- /dev/null +++ b/docs/LINK_TRACKING.md @@ -0,0 +1,116 @@ +# Link tracking and clean Sent copies (v9.4) + +Postmaster v9.4 adds per-link click telemetry to the existing per-recipient analytics pipeline. The existing open-tracking pixel remains unchanged: its route, event format, enrichment, client fingerprint, country/source, browser/OS parsing, campaign/delivery correlation and dashboard continue to work as before. + +## Recipient versus Sent architecture + +```text +canonical message + | + +-----------------------------+ + | | + v v +recipient variant Sent variant +existing tracking pixel no recipient pixel +HTTP/HTTPS -> /t/c/ original HTTP/HTTPS URLs +recipient AMP callback no recipient AMP callback part + | | + +-------- MIME build ----------+ +``` + +Tracking instrumentation belongs to the recipient delivery, not to the sender's archived Sent copy. v9.4 does not rewrite historical Sent messages. + +Link tracking follows the existing `track_opens` tracking opt-in (including the account default used by existing send/reply tools), so v9.4 does not add a second required send parameter. + +## Public click endpoint + +The only new public callback path required by v9.4 is: + +```text +GET /t/c/ +``` + +The public base is resolved exactly like existing mail callbacks: `PUBLIC_EMAIL_BASE_URL`, otherwise `https://`. + +The token is random and opaque; it does not encode recipient data or the destination. `/t/c/` resolves a server-side `tracking_links` record, records a `link` event and issues a `302` to the stored `original_url`. Request/query parameters are never used as the destination, so `?url=https://...` cannot create an open redirect. Invalid tokens return `404` without a destination hint. + +## HTML rewriting + +Recipient HTML rewrites eligible `http://` and `https://` anchors only. Postmaster does not rewrite `mailto:`, `tel:`, `cid:`, `data:`, `javascript:`, `#fragment` or other non-web URLs. A URL already pointing to this deployment's `/t/c/` path is not wrapped twice. A legitimate third-party URL whose path happens to contain `/t/c/` remains eligible. + +The original href is stored after normal HTML entity decoding, preserving query and fragment semantics. For example `https://example.com/page?a=1&b=2#section` is stored/redirected as `https://example.com/page?a=1&b=2#section`. Visible anchor text is unchanged. The original anchor index is retained, so repeated header/footer occurrences remain distinguishable while `normalized_url` supports destination aggregation. + +## Additive analytics schema + +Existing pixel tables are not replaced. v9.4 adds: + +`tracking_links`: `id`, `link_id`, random `tracking_token`, `campaign_id`, `delivery_id`, `account_id`, `recipient`, `message_id`, `original_url`, `normalized_url`, `destination_host`, `position`, `anchor_text`, `created_at`. + +`tracking_clicks`: `id`, `link_occurrence_id`, `link_id`, `delivery_id`, `campaign_id`, `account_id`, `recipient`, `observed_at`, `event_type=link`, `user_agent`, `client_fingerprint`, `country_code`, `browser`, `os`, `client_source`, `metadata_confidence`. + +The click fingerprint reuses the existing analytics HMAC derivation; raw IP is not stored. Country/source/browser/OS parsing reuses the pixel enrichment helpers. The browser field continues to contain the parsed browser/version label used by the existing pipeline. + +## Unique click + +A v9.4 unique click is: + +```text +delivery_id + link_id + client_fingerprint +``` + +If both IP and User-Agent are unavailable, the existing keyed fingerprint pipeline produces the stable HMAC of the empty pair; repeated unknown fetches for that delivery/link therefore collapse consistently. v9.4 does not classify human/bot/scanner clicks. User-Agent, source and fingerprint are retained for a future evidence-based classifier. + +## Analytics, MCP and dashboard + +Available analytics include total clicks, unique clicks, unique recipients, first/last click, campaign/delivery/link filtering, event detail, top links and destination host. + +Existing `tracking_status` and `get_tracking_campaign` are extended. v9.4 adds read-only `get_tracking_summary`, `list_tracking_links` and `list_tracking_events`. Listing/event tools never return opaque click tokens. + +The existing pixel dashboard is preserved. v9.4 adds **Top links** and a unified **Tracking events** table distinguishing `pixel`, `amp_xhr` and `link`; link rows include label, `link_id`, destination host and position. + +## Clean Sent behavior and historical finding + +Pre-v9.4, the individualized `EmailMessage` sent via SMTP was also serialized directly for IMAP APPEND to Sent. Therefore a tracked Sent copy could contain the recipient pixel and create a false self-open when the sender viewed it. + +v9.4 builds outbound and Sent MIME independently from the same canonical body/attachment inputs. Outbound keeps the existing pixel and tracked URLs. Sent keeps original URLs, no active recipient pixel, no `/t/c/` and no recipient AMP callback alternative. `Message-ID` and `Date` are synchronized; `Subject`, `In-Reply-To` and `References` come from the same canonical inputs. Attachment specs are reused and regression-tested for byte identity. No regex rewriting is performed on a serialized MIME blob. + +This is the primary self-open/self-click defense; Postmaster does not guess sender identity from IP, country, User-Agent, browser or fingerprint. + +## Cloudflare Access — manual deployment requirement + +Keep Postmaster protected by default. Existing public callback paths remain: + +```text +/track/open/* +/api/amp/* +``` + +v9.4 adds exactly one new required public path: + +```text +/t/c/* +``` + +**Cloudflare Access must bypass `/t/c/*`.** Without this bypass a recipient reaches the Access login/challenge instead of Postmaster and the redirect fails. + +Keep `/mcp`, dashboard/admin/private APIs, mail/task/memory/skill/file-management routes and tracking analytics protected. Do not disable Access globally. + +### Separate v9.3 file-handoff note + +`/files/{file_id}` is a pre-existing v9.3 signed HTTP file-handoff route. v9.4 does not add `/files/*` to its Cloudflare bypass or change that policy. If a deployment intentionally uses signed HTTP file handoff, the operator must separately ensure that pre-existing route is reachable according to the deployment's proxy policy. + +## Live preflight + +After release/deploy and after the operator adds the `/t/c/*` bypass: + +1. `build_status` reports `9.4.0`, `link_tracking=true`, `sent_copy_tracking_sanitized=true`. +2. Send a tracked email with at least two distinct HTTP/HTTPS URLs. +3. Recipient MIME contains the unchanged `/track/open/...gif` pixel and distinct `/t/c/` URLs. +4. Sent MIME contains neither recipient pixel nor `/t/c/`, and contains the original URLs. +5. Anonymous `GET https:///t/c/` reaches Postmaster without Cloudflare login, records a `link` event and redirects to the exact stored destination. +6. Invalid token plus `?url=https://evil.example/` does not redirect. +7. Link #1/#2 are independently visible in total/unique/top-link analytics with enrichment fields. +8. Clicking a URL extracted from Sent goes directly to the original destination and creates no Postmaster click event. +9. `/mcp`, dashboard and all other private paths remain protected by Cloudflare Access. + +The container cannot prove Cloudflare policy by itself. v9.4 is fully operational only after the external `/t/c/*` bypass is configured and the anonymous live redirect test succeeds. diff --git a/src/postmaster/link_tracking.py b/src/postmaster/link_tracking.py new file mode 100644 index 0000000..475403e --- /dev/null +++ b/src/postmaster/link_tracking.py @@ -0,0 +1,217 @@ +from __future__ import annotations + +import hashlib +import hmac +from functools import lru_cache +from typing import Any +from urllib.parse import urlsplit + +from .email_analytics import ( + AnalyticsError, EmailAnalyticsStore, _client_metadata, _now, _safe_base_url, _token, analytics_store, +) +from .link_tracking_queries import LinkTrackingQueriesMixin +from .link_tracking_html import ( + already_tracked_url, collect_anchors, eligible_web_url, normalized_url, replace_href, rewrite_anchor_tags, +) + +class LinkTrackingStore(LinkTrackingQueriesMixin): + """Additive per-link analytics layered on the existing email analytics database.""" + + def __init__(self, analytics: EmailAnalyticsStore | None = None): + self.analytics = analytics or analytics_store() + self._init_schema() + + def _connect(self): + return self.analytics._connect() + + def _init_schema(self) -> None: + with self._connect() as conn: + conn.executescript( + """ + CREATE TABLE IF NOT EXISTS tracking_links ( + id TEXT PRIMARY KEY, + link_id TEXT NOT NULL, + tracking_token TEXT NOT NULL UNIQUE, + campaign_id TEXT NOT NULL, + delivery_id TEXT NOT NULL, + account_id TEXT NOT NULL, + recipient TEXT NOT NULL COLLATE NOCASE, + original_url TEXT NOT NULL, + normalized_url TEXT NOT NULL, + destination_host TEXT NOT NULL DEFAULT '', + position INTEGER NOT NULL, + anchor_text TEXT NOT NULL DEFAULT '', + message_id TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL, + FOREIGN KEY(delivery_id) REFERENCES tracking_deliveries(id) ON DELETE CASCADE + ); + + CREATE INDEX IF NOT EXISTS ix_tracking_links_campaign + ON tracking_links(campaign_id, link_id); + CREATE INDEX IF NOT EXISTS ix_tracking_links_delivery + ON tracking_links(delivery_id, position); + CREATE INDEX IF NOT EXISTS ix_tracking_links_logical + ON tracking_links(link_id, campaign_id); + + CREATE TABLE IF NOT EXISTS tracking_clicks ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + link_occurrence_id TEXT NOT NULL, + link_id TEXT NOT NULL, + delivery_id TEXT NOT NULL, + campaign_id TEXT NOT NULL, + account_id TEXT NOT NULL, + recipient TEXT NOT NULL COLLATE NOCASE, + observed_at TEXT NOT NULL, + event_type TEXT NOT NULL DEFAULT 'link', + user_agent TEXT NOT NULL DEFAULT '', + client_fingerprint TEXT NOT NULL DEFAULT '', + country_code TEXT NOT NULL DEFAULT '', + browser TEXT NOT NULL DEFAULT '', + os TEXT NOT NULL DEFAULT '', + client_source TEXT NOT NULL DEFAULT '', + metadata_confidence TEXT NOT NULL DEFAULT '', + FOREIGN KEY(link_occurrence_id) REFERENCES tracking_links(id) ON DELETE CASCADE + ); + + CREATE INDEX IF NOT EXISTS ix_tracking_clicks_link + ON tracking_clicks(link_id, observed_at); + CREATE INDEX IF NOT EXISTS ix_tracking_clicks_delivery + ON tracking_clicks(delivery_id, observed_at); + CREATE INDEX IF NOT EXISTS ix_tracking_clicks_campaign + ON tracking_clicks(campaign_id, observed_at); + CREATE INDEX IF NOT EXISTS ix_tracking_clicks_recipient + ON tracking_clicks(recipient COLLATE NOCASE, observed_at); + """ + ) + cols = {str(row["name"]) for row in conn.execute("PRAGMA table_info(tracking_links)").fetchall()} + if "message_id" not in cols: + conn.execute( + "ALTER TABLE tracking_links ADD COLUMN message_id TEXT NOT NULL DEFAULT ''" + ) + + @staticmethod + def _logical_link_id(campaign_id: str, position: int, normalized_url: str, anchor_text: str) -> str: + digest = hashlib.sha256( + f"{campaign_id}\0{position}\0{normalized_url}\0{anchor_text}".encode("utf-8", "ignore") + ).hexdigest()[:20] + return f"lnk_{digest}" + + def _insert_link(self, *, delivery: dict[str, Any], original_url: str, position: int, anchor_text: str) -> dict[str, Any]: + normalized = normalized_url(original_url) + parts = urlsplit(original_url) + logical_id = self._logical_link_id(str(delivery["campaign_id"]), position, normalized, anchor_text) + occurrence_id = f"lno_{_token(12)}" + token = _token(24) + now = _now() + with self._connect() as conn: + conn.execute( + """ + INSERT INTO tracking_links( + id,link_id,tracking_token,campaign_id,delivery_id,account_id,recipient, + original_url,normalized_url,destination_host,position,anchor_text,message_id,created_at + ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?) + """, + ( + occurrence_id, logical_id, token, delivery["campaign_id"], delivery["id"], + delivery["account_id"], delivery["recipient"], original_url, normalized, + (parts.hostname or "").lower(), int(position), anchor_text, + str(delivery.get("message_id") or ""), now, + ), + ) + return { + "occurrence_id": occurrence_id, "link_id": logical_id, "tracking_token": token, + "campaign_id": str(delivery["campaign_id"]), "delivery_id": str(delivery["id"]), + "recipient": str(delivery["recipient"]), "original_url": original_url, + "normalized_url": normalized, "destination_host": (parts.hostname or "").lower(), + "position": int(position), "anchor_text": anchor_text, + "message_id": str(delivery.get("message_id") or ""), "created_at": now, + } + + def mark_delivery_message(self, delivery_id: str, message_id: str) -> None: + with self._connect() as conn: + conn.execute("UPDATE tracking_links SET message_id=? WHERE delivery_id=?", (message_id or "", delivery_id)) + + def instrument_html(self, *, body_html: str, delivery: dict[str, Any]) -> tuple[str, list[dict[str, Any]]]: + html = body_html or "" + anchors = collect_anchors(html) + if not anchors: + return html, [] + replacements: list[tuple[str, str]] = [] + public_base = _safe_base_url() + tracked: list[dict[str, Any]] = [] + for anchor_index, anchor in enumerate(anchors): + href = str(anchor.get("href") or "") + if not eligible_web_url(href) or already_tracked_url(href, public_base): + continue + record = self._insert_link( + delivery=delivery, original_url=href, position=anchor_index, + anchor_text=str(anchor.get("anchor_text") or "")[:500], + ) + tracked_url = f"{public_base}/t/c/{record['tracking_token']}" + replacements.append((str(anchor["raw_tag"]), replace_href(str(anchor["raw_tag"]), tracked_url))) + safe = dict(record) + safe.pop("tracking_token", None) + safe["tracked_url_path"] = "/t/c/" + tracked.append(safe) + return rewrite_anchor_tags(html, replacements), tracked + + def get_by_token(self, token: str) -> dict[str, Any]: + with self._connect() as conn: + row = conn.execute("SELECT * FROM tracking_links WHERE tracking_token=?", (token,)).fetchone() + if not row: + raise AnalyticsError("Unknown link tracking token") + return dict(row) + + def _fingerprint(self, client_ip: str, user_agent: str) -> str: + return hmac.new( + self.analytics._fingerprint_key, + f"{client_ip}|{user_agent}".encode("utf-8", "ignore"), + hashlib.sha256, + ).hexdigest()[:24] + + def record_click(self, link: dict[str, Any], *, user_agent: str = "", client_ip: str = "", country_code: str = "") -> dict[str, Any]: + now = _now() + ua = (user_agent or "")[:500] + metadata = _client_metadata("link", ua, country_code) + fingerprint = self._fingerprint(client_ip or "", ua) + with self._connect() as conn: + cur = conn.execute( + """ + INSERT INTO tracking_clicks( + link_occurrence_id,link_id,delivery_id,campaign_id,account_id,recipient, + observed_at,event_type,user_agent,client_fingerprint,country_code,browser,os, + client_source,metadata_confidence + ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + """, + ( + link["id"], link["link_id"], link["delivery_id"], link["campaign_id"], + link["account_id"], link["recipient"], now, "link", ua, fingerprint, + metadata["country_code"], metadata["browser"], metadata["os"], + metadata["client_source"], metadata["metadata_confidence"], + ), + ) + event_id = int(cur.lastrowid) + return { + "ok": True, "id": event_id, "event_type": "link", "link_id": str(link["link_id"]), + "delivery_id": str(link["delivery_id"]), "campaign_id": str(link["campaign_id"]), + "recipient": str(link["recipient"]), "observed_at": now, + "client_fingerprint": fingerprint, "country_code": metadata["country_code"], + "browser": metadata["browser"], "os": metadata["os"], + "client_source": metadata["client_source"], + "metadata_confidence": metadata["metadata_confidence"], + } + + def status(self) -> dict[str, Any]: + with self._connect() as conn: + links = conn.execute("SELECT COUNT(*) FROM tracking_links").fetchone()[0] + clicks = conn.execute("SELECT COUNT(*) FROM tracking_clicks").fetchone()[0] + return { + "link_tracking": True, "link_occurrences": int(links), "click_events": int(clicks), + "public_path": "/t/c/*", "unique_click_definition": "delivery_id + link_id + client_fingerprint", + "tokens_exposed_in_listing": False, + } + + +@lru_cache(maxsize=1) +def link_store() -> LinkTrackingStore: + return LinkTrackingStore(analytics_store()) diff --git a/src/postmaster/link_tracking_html.py b/src/postmaster/link_tracking_html.py new file mode 100644 index 0000000..854902c --- /dev/null +++ b/src/postmaster/link_tracking_html.py @@ -0,0 +1,117 @@ +from __future__ import annotations + +import re +from html import escape, unescape +from html.parser import HTMLParser +from typing import Any +from urllib.parse import urlsplit, urlunsplit + +_HREF_RE = re.compile(r"(?is)\bhref\s*=\s*(?:\"([^\"]*)\"|'([^']*)'|([^\s>]+))") + + +def _href_value(raw_tag: str): + match = _HREF_RE.search(raw_tag or "") + if not match: + return None, None + for group in (1, 2, 3): + value = match.group(group) + if value is not None: + return unescape(value).strip(), match + return None, None + + +def replace_href(raw_tag: str, new_url: str) -> str: + _, match = _href_value(raw_tag) + if not match: + return raw_tag + for group in (1, 2, 3): + if match.group(group) is not None: + start, end = match.span(group) + return raw_tag[:start] + escape(new_url, quote=True) + raw_tag[end:] + return raw_tag + + +def eligible_web_url(url: str) -> bool: + try: + parts = urlsplit((url or "").strip()) + except ValueError: + return False + return parts.scheme.lower() in {"http", "https"} and bool(parts.netloc) + + +def already_tracked_url(url: str, public_base: str) -> bool: + try: + target, base = urlsplit(url), urlsplit(public_base) + except ValueError: + return False + return ( + target.scheme.lower() == base.scheme.lower() + and target.netloc.lower() == base.netloc.lower() + and target.path.startswith("/t/c/") + ) + + +def normalized_url(url: str) -> str: + parts = urlsplit(url) + return urlunsplit((parts.scheme.lower(), parts.netloc.lower(), parts.path, parts.query, parts.fragment)) + + +class _AnchorCollector(HTMLParser): + def __init__(self) -> None: + super().__init__(convert_charrefs=False) + self.anchors: list[dict[str, Any]] = [] + self._active: list[int] = [] + + def handle_starttag(self, tag: str, attrs) -> None: + raw = self.get_starttag_text() or "" + if tag.casefold() == "a": + href, _ = _href_value(raw) + self.anchors.append({"raw_tag": raw, "href": href or "", "text": []}) + self._active.append(len(self.anchors) - 1) + elif tag.casefold() == "img" and self._active: + alt = next((str(v or "") for k, v in attrs if str(k).casefold() == "alt"), "") + if alt: + self.anchors[self._active[-1]]["text"].append(alt) + + def handle_endtag(self, tag: str) -> None: + if tag.casefold() == "a" and self._active: + self._active.pop() + + def handle_data(self, data: str) -> None: + if self._active: + self.anchors[self._active[-1]]["text"].append(data) + + def handle_entityref(self, name: str) -> None: + if self._active: + self.anchors[self._active[-1]]["text"].append(unescape(f"&{name};")) + + def handle_charref(self, name: str) -> None: + if self._active: + self.anchors[self._active[-1]]["text"].append(unescape(f"&#{name};")) + + +def collect_anchors(html: str) -> list[dict[str, Any]]: + parser = _AnchorCollector() + try: + parser.feed(html or "") + parser.close() + except Exception: + return [] + for anchor in parser.anchors: + anchor["anchor_text"] = re.sub(r"\s+", " ", "".join(anchor.pop("text", []))).strip() + return parser.anchors + + +def rewrite_anchor_tags(html: str, replacements: list[tuple[str, str]]) -> str: + cursor = 0 + pieces: list[str] = [] + for old_tag, new_tag in replacements: + pos = html.find(old_tag, cursor) + if pos < 0: + continue + pieces.extend((html[cursor:pos], new_tag)) + cursor = pos + len(old_tag) + if not pieces: + return html + pieces.append(html[cursor:]) + return "".join(pieces) diff --git a/src/postmaster/link_tracking_queries.py b/src/postmaster/link_tracking_queries.py new file mode 100644 index 0000000..72e026c --- /dev/null +++ b/src/postmaster/link_tracking_queries.py @@ -0,0 +1,230 @@ +from __future__ import annotations + +from typing import Any + + +class LinkTrackingQueriesMixin: + def list_links( + self, + *, + campaign_id: str | None = None, + delivery_id: str | None = None, + link_id: str | None = None, + account_id: str | None = None, + clicked_only: bool = False, + limit: int = 500, + ) -> list[dict[str, Any]]: + clauses: list[str] = [] + params: list[Any] = [] + for column, value in ( + ("l.campaign_id", campaign_id), + ("l.delivery_id", delivery_id), + ("l.link_id", link_id), + ("l.account_id", account_id), + ): + if value: + clauses.append(f"{column}=?") + params.append(value) + if clicked_only: + clauses.append("EXISTS (SELECT 1 FROM tracking_clicks cx WHERE cx.link_occurrence_id=l.id)") + where = (" WHERE " + " AND ".join(clauses)) if clauses else "" + limit = max(1, min(int(limit), 2000)) + params.append(limit) + with self._connect() as conn: + rows = conn.execute( + f""" + SELECT l.id AS occurrence_id,l.link_id,l.campaign_id,l.delivery_id,l.account_id, + l.recipient,l.original_url,l.normalized_url,l.destination_host,l.position, + l.anchor_text,l.message_id,l.created_at, + COUNT(c.id) AS total_clicks, + COUNT(DISTINCT c.client_fingerprint) AS unique_clicks, + MIN(c.observed_at) AS first_click, + MAX(c.observed_at) AS last_click + FROM tracking_links l + LEFT JOIN tracking_clicks c ON c.link_occurrence_id=l.id + {where} + GROUP BY l.id + ORDER BY l.created_at DESC,l.position ASC + LIMIT ? + """, + params, + ).fetchall() + return [dict(row) for row in rows] + + def list_click_events( + self, + *, + campaign_id: str | None = None, + delivery_id: str | None = None, + link_id: str | None = None, + recipient: str | None = None, + account_id: str | None = None, + limit: int = 500, + ) -> list[dict[str, Any]]: + clauses: list[str] = [] + params: list[Any] = [] + for column, value in ( + ("c.campaign_id", campaign_id), + ("c.delivery_id", delivery_id), + ("c.link_id", link_id), + ("c.account_id", account_id), + ): + if value: + clauses.append(f"{column}=?") + params.append(value) + if recipient: + clauses.append("c.recipient=? COLLATE NOCASE") + params.append(recipient.strip()) + where = (" WHERE " + " AND ".join(clauses)) if clauses else "" + limit = max(1, min(int(limit), 2000)) + params.append(limit) + with self._connect() as conn: + rows = conn.execute( + f""" + SELECT c.*,l.original_url,l.normalized_url,l.destination_host,l.position,l.anchor_text + FROM tracking_clicks c + JOIN tracking_links l ON l.id=c.link_occurrence_id + {where} + ORDER BY c.observed_at DESC,c.id DESC + LIMIT ? + """, + params, + ).fetchall() + return [dict(row) for row in rows] + + def summary( + self, + *, + campaign_id: str | None = None, + delivery_id: str | None = None, + link_id: str | None = None, + account_id: str | None = None, + ) -> dict[str, Any]: + clauses: list[str] = [] + params: list[Any] = [] + for column, value in ( + ("l.campaign_id", campaign_id), + ("l.delivery_id", delivery_id), + ("l.link_id", link_id), + ("l.account_id", account_id), + ): + if value: + clauses.append(f"{column}=?") + params.append(value) + where = (" WHERE " + " AND ".join(clauses)) if clauses else "" + with self._connect() as conn: + row = conn.execute( + f""" + SELECT COUNT(DISTINCT l.id) AS link_occurrences, + COUNT(DISTINCT l.link_id) AS logical_links, + COUNT(c.id) AS total_clicks, + COUNT(DISTINCT c.delivery_id || '|' || c.link_id || '|' || c.client_fingerprint) AS unique_clicks, + COUNT(DISTINCT CASE WHEN c.id IS NOT NULL THEN c.recipient END) AS unique_recipients, + MIN(c.observed_at) AS first_click, + MAX(c.observed_at) AS last_click + FROM tracking_links l + LEFT JOIN tracking_clicks c ON c.link_occurrence_id=l.id + {where} + """, + params, + ).fetchone() + out = dict(row) + out["unique_click_definition"] = "delivery_id + link_id + client_fingerprint" + out["fingerprint_fallback"] = ( + "If IP and User-Agent are unavailable, the existing keyed fingerprint pipeline " + "uses the stable HMAC of the empty pair, collapsing unknown repeat fetches for that delivery/link." + ) + return out + + def top_links( + self, + *, + campaign_id: str | None = None, + delivery_id: str | None = None, + account_id: str | None = None, + limit: int = 25, + ) -> list[dict[str, Any]]: + clauses: list[str] = [] + params: list[Any] = [] + for column, value in ( + ("l.campaign_id", campaign_id), + ("l.delivery_id", delivery_id), + ("l.account_id", account_id), + ): + if value: + clauses.append(f"{column}=?") + params.append(value) + where = (" WHERE " + " AND ".join(clauses)) if clauses else "" + limit = max(1, min(int(limit), 200)) + params.append(limit) + with self._connect() as conn: + rows = conn.execute( + f""" + SELECT l.link_id, + MIN(l.anchor_text) AS anchor_text, + MIN(l.original_url) AS original_url, + MIN(l.normalized_url) AS normalized_url, + MIN(l.destination_host) AS destination_host, + MIN(l.position) AS position, + COUNT(c.id) AS total_clicks, + COUNT(DISTINCT c.delivery_id || '|' || c.link_id || '|' || c.client_fingerprint) AS unique_clicks, + COUNT(DISTINCT CASE WHEN c.id IS NOT NULL THEN c.recipient END) AS unique_recipients, + MIN(c.observed_at) AS first_click, + MAX(c.observed_at) AS last_click + FROM tracking_links l + LEFT JOIN tracking_clicks c ON c.link_occurrence_id=l.id + {where} + GROUP BY l.link_id + ORDER BY total_clicks DESC,unique_clicks DESC,l.link_id + LIMIT ? + """, + params, + ).fetchall() + return [dict(row) for row in rows] + + def unified_events( + self, + *, + campaign_id: str | None = None, + delivery_id: str | None = None, + link_id: str | None = None, + recipient: str | None = None, + account_id: str | None = None, + event_type: str | None = None, + limit: int = 500, + ) -> list[dict[str, Any]]: + want = (event_type or "all").strip().lower() + rows: list[dict[str, Any]] = [] + if want in {"all", "pixel", "amp_xhr"} and not link_id: + opens = self.analytics.list_open_events( + delivery_id=delivery_id, + campaign_id=campaign_id, + recipient=recipient, + account_id=account_id, + limit=limit, + ) + for event in opens: + if want != "all" and str(event.get("event_type")) != want: + continue + rows.append({ + **event, + "observed_at": event.get("opened_at", ""), + "link_id": "", + "anchor_text": "", + "original_url": "", + "destination_host": "", + "position": None, + }) + if want in {"all", "link"}: + rows.extend( + self.list_click_events( + campaign_id=campaign_id, + delivery_id=delivery_id, + link_id=link_id, + recipient=recipient, + account_id=account_id, + limit=limit, + ) + ) + rows.sort(key=lambda item: str(item.get("observed_at") or ""), reverse=True) + return rows[: max(1, min(int(limit), 2000))] diff --git a/src/postmaster/runtime.py b/src/postmaster/runtime.py index 5c2f0c4..bf79bbd 100644 --- a/src/postmaster/runtime.py +++ b/src/postmaster/runtime.py @@ -1,10 +1,12 @@ from __future__ import annotations import os +from html import escape import uvicorn from mcp.types import CallToolResult from starlette.requests import Request +from starlette.responses import HTMLResponse, PlainTextResponse, RedirectResponse from starlette.routing import Mount, Route from . import server as _base @@ -13,21 +15,30 @@ stored_file_http_response, stored_file_resource_result, ) - +from .link_tracking import link_store +from .link_tracking_html import eligible_web_url +from .tracked_mail import LinkTrackingMailClient mcp = _base.mcp _legacy_build_status = _base.build_status +_legacy_tracking_status = _base.tracking_status +_legacy_get_tracking_campaign = _base.get_tracking_campaign +_legacy_dashboard_home = _base.dashboard_home + + +def mail_client(account_id: str | None = None) -> LinkTrackingMailClient: + return LinkTrackingMailClient(_base.account_store().settings(account_id)) + +_base.mail_client = mail_client def build_status(): - """Read-only. Return running build identity and v9.3 file-handoff capability.""" status = _legacy_build_status() status["native_file_resource_handoff"] = True + status["link_tracking"] = True + status["sent_copy_tracking_sanitized"] = True return status - -# Replace only the registered build-status implementation. MCPServer v2 exposes -# remove_tool() and add_tool() as public APIs; all v9.2 upload/file/mail tools stay intact. mcp.remove_tool("build_status") mcp.add_tool(build_status, name="build_status") _base.build_status = build_status @@ -35,14 +46,6 @@ def build_status(): @mcp.tool() def get_stored_file_resource(file_id: str, transport: str = "auto") -> CallToolResult: - """ - Read-only. Return a native MCP ResourceLink for a FileStore file. - - transport=auto prefers a configured signed HTTPS file URL and otherwise - returns the canonical postmaster:// resource URI. transport=http requires - FILE_STORE_PUBLIC_BASE_URL or PUBLIC_MCP_HOST; transport=mcp always returns - the MCP resource. - """ return stored_file_resource_result(_base.file_store(), file_id, transport) @@ -52,10 +55,86 @@ def get_stored_file_resource(file_id: str, transport: str = "auto") -> CallToolR description="Original bytes for a Postmaster FileStore file identified by canonical file_id.", ) def stored_file_resource(file_id: str) -> bytes: - """Read original FileStore bytes; the MCP SDK emits BlobResourceContents for binary bytes.""" return read_stored_file_resource(_base.file_store(), file_id) +def tracking_status(): + base = _legacy_tracking_status() + if isinstance(base, dict) and base.get("ok"): + base["link_tracking"] = link_store().status() + base["event_types"] = ["pixel", "amp_xhr", "link"] + return base + +mcp.remove_tool("tracking_status") +mcp.add_tool(tracking_status, name="tracking_status") +_base.tracking_status = tracking_status + + +def get_tracking_campaign(campaign_id: str): + base = _legacy_get_tracking_campaign(campaign_id) + if isinstance(base, dict) and base.get("ok") is False: + return base + try: + base["link_tracking"] = link_store().summary(campaign_id=campaign_id) + base["top_links"] = link_store().top_links(campaign_id=campaign_id, limit=25) + except Exception as exc: + base["link_tracking_error"] = f"{type(exc).__name__}: {exc}" + return base + +mcp.remove_tool("get_tracking_campaign") +mcp.add_tool(get_tracking_campaign, name="get_tracking_campaign") +_base.get_tracking_campaign = get_tracking_campaign + + +@mcp.tool() +def get_tracking_summary( + campaign_id: str | None = None, + delivery_id: str | None = None, + link_id: str | None = None, + account_id: str | None = None, +): + """Read-only click summary. Unique click = delivery_id + link_id + client_fingerprint.""" + return _base._safe_call( + link_store().summary, + campaign_id=campaign_id, delivery_id=delivery_id, link_id=link_id, account_id=account_id, + ) + + +@mcp.tool() +def list_tracking_links( + campaign_id: str | None = None, + delivery_id: str | None = None, + link_id: str | None = None, + account_id: str | None = None, + clicked_only: bool = False, + limit: int = 500, +): + """Read-only tracked link occurrences and aggregates; opaque tokens are never returned.""" + return _base._safe_call( + link_store().list_links, + campaign_id=campaign_id, delivery_id=delivery_id, link_id=link_id, + account_id=account_id, clicked_only=clicked_only, limit=limit, + ) + + +@mcp.tool() +def list_tracking_events( + delivery_id: str | None = None, + campaign_id: str | None = None, + link_id: str | None = None, + recipient: str | None = None, + account_id: str | None = None, + event_type: str | None = None, + limit: int = 500, +): + """Read-only unified tracking events. event_type may be all, pixel, amp_xhr or link.""" + return _base._safe_call( + link_store().unified_events, + delivery_id=delivery_id, campaign_id=campaign_id, link_id=link_id, + recipient=recipient, account_id=account_id, event_type=event_type, limit=limit, + ) + + async def public_stored_file_download(request: Request): return stored_file_http_response(request, _base.file_store(), require_signature=True) @@ -64,40 +143,131 @@ async def dashboard_file_download(request: Request): return stored_file_http_response(request, _base.file_store(), require_signature=False) -# Replace the old dashboard download endpoint (which materialized the whole blob) -# and insert the signed handoff route before the catch-all MCP Mount. +async def tracking_click(request: Request): + token = str(request.path_params.get("token", "")) + try: + link = link_store().get_by_token(token) + destination = str(link.get("original_url") or "") + if not eligible_web_url(destination): + raise ValueError("Stored link destination is not an HTTP/HTTPS URL") + except Exception: + return PlainTextResponse("Not found", status_code=404, headers={"Cache-Control": "no-store"}) + + forwarded = request.headers.get("x-forwarded-for", "") + client_ip = forwarded.split(",", 1)[0].strip() if forwarded else "" + if not client_ip and request.client: + client_ip = request.client.host or "" + try: + link_store().record_click( + link, + user_agent=request.headers.get("user-agent", ""), + client_ip=client_ip, + country_code=request.headers.get("cf-ipcountry", ""), + ) + except Exception: + _base.logger.info("Link click could not be recorded", exc_info=True) + + response = RedirectResponse(destination, status_code=302) + response.headers["Cache-Control"] = "private, no-store, no-cache, max-age=0" + response.headers["Pragma"] = "no-cache" + return response + + +def _tracking_dashboard_fragment(account_id: str | None = None) -> str: + top = link_store().top_links(account_id=account_id, limit=20) + events = link_store().unified_events(account_id=account_id, limit=100) + top_rows = [] + for row in top: + label = str(row.get("anchor_text") or row.get("destination_host") or row.get("original_url") or "") + top_rows.append( + "" + f"{escape(str(row.get('link_id','')))}" + f"{escape(label)}" + f"{escape(str(row.get('destination_host','')))}" + f"{int(row.get('total_clicks') or 0)}" + f"{int(row.get('unique_clicks') or 0)}" + f"{int(row.get('unique_recipients') or 0)}" + f"{escape(str(row.get('first_click') or ''))}" + f"{escape(str(row.get('last_click') or ''))}" + ) + if not top_rows: + top_rows.append('No link clicks recorded yet.') + + event_rows = [] + for row in events: + source = " / ".join(x for x in (str(row.get("country_code") or ""), str(row.get("client_source") or "")) if x) + browser_os = " / ".join(x for x in (str(row.get("browser") or ""), str(row.get("os") or "")) if x) + label = str(row.get("anchor_text") or row.get("destination_host") or "") + ua = str(row.get("user_agent") or "")[:180] + event_rows.append( + "" + f"{escape(str(row.get('event_type') or ''))}" + f"{escape(str(row.get('recipient') or ''))}" + f"{escape(str(row.get('observed_at') or ''))}" + f"{escape(source)}{escape(browser_os)}" + f"{escape(str(row.get('campaign_id') or ''))}
{escape(str(row.get('delivery_id') or ''))}" + f"{escape(str(row.get('client_fingerprint') or ''))}" + f"{escape(label)}{escape(str(row.get('link_id') or ''))}" + f"{escape(str(row.get('destination_host') or ''))}" + f"{escape(str(row.get('position') if row.get('position') is not None else ''))}" + f"{escape(ua)}" + ) + if not event_rows: + event_rows.append('No tracking events recorded yet.') + + return f""" +
+

Top links

v9.4 click analytics
+

Unique click = delivery_id + link_id + client_fingerprint. Fetches are telemetry; v9.4 does not classify human vs scanner.

+
{''.join(top_rows)}
Link IDLabelDestination hostTotalUniqueRecipientsFirst clickLast click
+
+
+

Tracking events

pixel / AMP / link
+
{''.join(event_rows)}
TypeRecipientObserved UTCCountry / sourceBrowser / OSCampaign / deliveryClient fingerprintLink labelLink IDDestinationPositionUser-Agent
+
+""" + + +async def dashboard_home(request: Request): + response = await _legacy_dashboard_home(request) + if "text/html" not in str(response.headers.get("content-type", "")).lower(): + return response + try: + body = response.body.decode("utf-8") + fragment = _tracking_dashboard_fragment(request.query_params.get("account") or None) + marker = '
\n

Accuracy / privacy model

' + if marker in body: + body = body.replace(marker, fragment + marker, 1) + elif "" in body: + body = body.replace("", fragment + "", 1) + else: + body += fragment + return HTMLResponse(body, status_code=response.status_code) + except Exception: + _base.logger.info("Could not augment tracking dashboard", exc_info=True) + return response + + _routes = _base.app.router.routes for index, route in enumerate(list(_routes)): if isinstance(route, Route) and route.path == "/dashboard/files/{file_id}/download": - _routes[index] = Route( - "/dashboard/files/{file_id}/download", - dashboard_file_download, - methods=["GET", "HEAD"], - ) - break + _routes[index] = Route("/dashboard/files/{file_id}/download", dashboard_file_download, methods=["GET", "HEAD"]) + elif isinstance(route, Route) and route.path == "/": + _routes[index] = Route("/", dashboard_home, methods=["GET"]) +mount_index = next((i for i, route in enumerate(_routes) if isinstance(route, Mount)), len(_routes)) +if not any(isinstance(route, Route) and route.path == "/t/c/{token}" for route in _routes): + _routes.insert(mount_index, Route("/t/c/{token}", tracking_click, methods=["GET"])) + mount_index += 1 if not any(isinstance(route, Route) and route.path == "/files/{file_id}" for route in _routes): - mount_index = next((i for i, route in enumerate(_routes) if isinstance(route, Mount)), len(_routes)) - _routes.insert( - mount_index, - Route("/files/{file_id}", public_stored_file_download, methods=["GET", "HEAD"]), - ) - + _routes.insert(mount_index, Route("/files/{file_id}", public_stored_file_download, methods=["GET", "HEAD"])) app = _base.app -# Re-export the v9.2 and earlier callable surface for tests/importers that use the -# deployment entrypoint module rather than postmaster.server directly. for _name in dir(_base): if _name.startswith("_") or _name in globals(): continue globals()[_name] = getattr(_base, _name) - if __name__ == "__main__": - uvicorn.run( - app, - host=os.getenv("MCP_HOST", "0.0.0.0"), - port=int(os.getenv("MCP_PORT", "8000")), - log_level="info", - ) + uvicorn.run(app, host=os.getenv("MCP_HOST", "0.0.0.0"), port=int(os.getenv("MCP_PORT", "8000")), log_level="info") diff --git a/src/postmaster/tracked_mail.py b/src/postmaster/tracked_mail.py new file mode 100644 index 0000000..c78c5a9 --- /dev/null +++ b/src/postmaster/tracked_mail.py @@ -0,0 +1,179 @@ +from __future__ import annotations + +import imaplib +import smtplib +import ssl +from datetime import datetime +from email import policy +from email.message import EmailMessage +from email.utils import format_datetime, make_msgid +from html import escape +from typing import Any + +from .email_analytics import analytics_store +from .link_tracking import link_store +from .mail_bridge import MailBridgeError +from .mail_extensions import EnhancedMailClient, _plain_to_html + +def _sent_clean_html(body_html: str, delivery: dict[str, Any]) -> str: + """Render recipient-visible placeholders without retaining recipient telemetry URLs.""" + html = body_html or "" + replacements = { + "{{RECIPIENT}}": str(delivery["recipient"]), + "{{CAMPAIGN_ID}}": str(delivery["campaign_id"]), + "{{DELIVERY_ID}}": str(delivery["id"]), + "{{AMP_STATUS_URL}}": "", + "{{TRACKING_PIXEL_URL}}": "", + } + for key, value in replacements.items(): + html = html.replace(key, escape(value, quote=True)) + return html + + +def _synchronize_transport_headers(outbound: EmailMessage, sent_copy: EmailMessage, sender: str) -> None: + if "Date" not in outbound: + outbound["Date"] = format_datetime(datetime.now().astimezone()) + if "Message-ID" not in outbound: + domain = sender.rsplit("@", 1)[-1] + outbound["Message-ID"] = make_msgid(domain=domain) + for header in ("Date", "Message-ID"): + if header in sent_copy: + del sent_copy[header] + sent_copy[header] = str(outbound[header]) + + +class LinkTrackingMailClient(EnhancedMailClient): + """v9.4 delivery variant: tracked recipient MIME plus clean archived Sent MIME.""" + + def _send_message_with_clean_sent(self, outbound: EmailMessage, sent_copy: EmailMessage, recipients: list[str]) -> dict[str, Any]: + if not self.settings.enable_send: + raise MailBridgeError( + "Sending is disabled. Set ENABLE_SEND=true only when you are ready to allow SMTP writes." + ) + _synchronize_transport_headers(outbound, sent_copy, self.settings.email_address) + context = ssl.create_default_context() + security = (self.settings.smtp_security or ("starttls" if self.settings.smtp_starttls else "ssl")).strip().lower() + username = self.settings.smtp_username or self.settings.email_address + password = self.settings.smtp_password or self.settings.email_password + if security == "ssl": + with smtplib.SMTP_SSL(self.settings.smtp_host, self.settings.smtp_port, timeout=30, context=context) as smtp: + smtp.login(username, password) + smtp.send_message(outbound, from_addr=self.settings.email_address, to_addrs=recipients) + else: + with smtplib.SMTP(self.settings.smtp_host, self.settings.smtp_port, timeout=30) as smtp: + smtp.ehlo() + if security == "starttls": + smtp.starttls(context=context) + smtp.ehlo() + elif security != "plain": + raise MailBridgeError(f"Unsupported SMTP security mode: {security}") + smtp.login(username, password) + smtp.send_message(outbound, from_addr=self.settings.email_address, to_addrs=recipients) + if hasattr(self, "_sent_addresses_cache"): + delattr(self, "_sent_addresses_cache") + sent_copy_saved = False + sent_copy_error = None + if self.settings.save_sent_copy: + try: + with self._imap() as conn: + typ, _ = conn.append( + self.settings.sent_mailbox, r"\Seen", + imaplib.Time2Internaldate(datetime.now().timestamp()), + sent_copy.as_bytes(policy=policy.SMTP), + ) + sent_copy_saved = typ == "OK" + if not sent_copy_saved: + sent_copy_error = "IMAP APPEND returned non-OK" + except Exception as exc: + sent_copy_error = type(exc).__name__ + return { + "sent": True, "from": self.settings.email_address, "to": recipients, + "subject": str(outbound.get("Subject", "")), "message_id": str(outbound.get("Message-ID", "")), + "sent_copy_saved": sent_copy_saved, "sent_copy_error": sent_copy_error, + "sent_copy_tracking_sanitized": True, + } + + def _send_individualized( + self, *, to: list[str], subject: str, body: str = "", cc: list[str] | None = None, + bcc: list[str] | None = None, body_html: str | None = None, body_amp: str | None = None, + attachments: list[dict[str, Any]] | None = None, track_opens: bool, + campaign_id: str | None = None, in_reply_to: str = "", references: str = "", + ) -> dict[str, Any]: + amp_used = bool(body_amp) + to_clean = self._validate_recipients(to) + cc_clean = self._validate_recipients(cc or []) if cc else [] + bcc_clean = self._validate_recipients(bcc or []) if bcc else [] + recipient_roles: list[tuple[str, str]] = [] + seen: set[str] = set() + for role, addresses in (("to", to_clean), ("cc", cc_clean), ("bcc", bcc_clean)): + for address in addresses: + key = address.lower() + if key not in seen: + seen.add(key) + recipient_roles.append((address, role)) + analytics = analytics_store() + links = link_store() + if track_opens or amp_used: + analytics.validate_public_base_url() + campaign = analytics.create_campaign( + account_id=getattr(self.settings, "account_id", "") or self.settings.email_address, + sender=self.settings.email_address, subject=subject.strip(), track_opens=track_opens, + amp_used=amp_used, campaign_id=campaign_id, + ) + base_html = body_html if body_html is not None else _plain_to_html(body) + delivery_results: list[dict[str, Any]] = [] + errors: list[dict[str, Any]] = [] + attachment_meta: list[dict[str, Any]] = [] + for recipient, role in recipient_roles: + delivery = analytics.create_delivery( + campaign_id=campaign["id"], + account_id=getattr(self.settings, "account_id", "") or self.settings.email_address, + recipient=recipient, recipient_role=role, + ) + recipient_html, recipient_amp = analytics.render_for_recipient( + body_html=base_html, body_amp=body_amp, delivery=delivery, track_opens=track_opens, + ) + link_meta: list[dict[str, Any]] = [] + if track_opens: + recipient_html, link_meta = links.instrument_html(body_html=recipient_html, delivery=delivery) + clean_html = _sent_clean_html(base_html, delivery) + try: + outbound, _, meta = self._build_message( + to=to_clean, cc=cc_clean, subject=subject, body=body, body_html=recipient_html, + body_amp=recipient_amp, attachments=attachments, allow_unlisted=False, + in_reply_to=in_reply_to, references=references, + ) + sent_copy, _, _ = self._build_message( + to=to_clean, cc=cc_clean, subject=subject, body=body, body_html=clean_html, + body_amp=None, attachments=attachments, allow_unlisted=False, + in_reply_to=in_reply_to, references=references, + ) + result = self._send_message_with_clean_sent(outbound, sent_copy, [recipient]) + analytics.mark_sent(delivery["id"], str(result.get("message_id", ""))) + links.mark_delivery_message(delivery["id"], str(result.get("message_id", ""))) + attachment_meta = meta + delivery_results.append({ + "delivery_id": delivery["id"], "recipient": recipient, "role": role, + "message_id": result.get("message_id", ""), + "sent_copy_saved": result.get("sent_copy_saved", False), + "sent_copy_tracking_sanitized": True, "link_tracking": bool(track_opens), + "links": link_meta, + }) + except Exception as exc: + errors.append({ + "delivery_id": delivery["id"], "recipient": recipient, "role": role, + "error": f"{type(exc).__name__}: {exc}", + }) + return { + "sent": bool(delivery_results) and not errors, "partial": bool(delivery_results) and bool(errors), + "from": self.settings.email_address, "subject": subject.strip(), "campaign_id": campaign["id"], + "individualized": True, "visible_recipient_headers_preserved": True, + "tracked": bool(track_opens), "link_tracking": bool(track_opens), + "sent_copy_tracking_sanitized": True, "amp": amp_used, + "amp_registered": bool(getattr(self.settings, "amp_registered", False)), + "deliveries": delivery_results, "errors": errors, "attachments": attachment_meta, + "tracking_note": ( + "Open and click events are fetch telemetry and may be affected by mail proxies, " + "security scanners or prefetching; v9.4 does not classify human vs scanner clicks." + ) if track_opens else "", + } diff --git a/tests/test_v9_4_link_tracking_store.py b/tests/test_v9_4_link_tracking_store.py new file mode 100644 index 0000000..d1314d6 --- /dev/null +++ b/tests/test_v9_4_link_tracking_store.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +import os +import tempfile +import unittest +from pathlib import Path + +from postmaster.email_analytics import EmailAnalyticsStore +from postmaster.link_tracking import LinkTrackingStore +from postmaster.tracked_mail import _sent_clean_html + +class LinkStoreTests(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + root = Path(self.tmp.name) + self.old_public = {k: os.environ.get(k) for k in ("PUBLIC_EMAIL_BASE_URL", "PUBLIC_MCP_HOST")} + os.environ["PUBLIC_EMAIL_BASE_URL"] = "https://postmaster.example.test" + os.environ["PUBLIC_MCP_HOST"] = "" + self.analytics = EmailAnalyticsStore(db_path=str(root / "analytics.db"), key_path=str(root / "analytics.key")) + self.links = LinkTrackingStore(self.analytics) + self.campaign = self.analytics.create_campaign(account_id="acct", sender="sender@example.test", subject="v9.4", track_opens=True, amp_used=False) + self.delivery = self.analytics.create_delivery(campaign_id=self.campaign["id"], account_id="acct", recipient="reader@example.test", recipient_role="to") + + def tearDown(self) -> None: + for key, value in self.old_public.items(): + if value is None: os.environ.pop(key, None) + else: os.environ[key] = value + self.tmp.cleanup() + + def _token_for_occurrence(self, occurrence_id: str) -> str: + with self.links._connect() as conn: + return str(conn.execute("SELECT tracking_token FROM tracking_links WHERE id=?", (occurrence_id,)).fetchone()[0]) + + def test_html_rewrite_rules_query_fragment_positions_and_no_double_wrap(self) -> None: + original = ( + '

HTTP One' + 'HTTPS Two' + 'MailTel' + 'CidAnchor' + 'DataJS' + 'Already tracked' + 'Third-party /t/c' + 'HTTPS Two footer

' + ) + rewritten, meta = self.links.instrument_html(body_html=original, delivery=self.delivery) + self.assertEqual(len(meta), 4) + self.assertEqual([row["position"] for row in meta], [0, 1, 9, 10]) + self.assertEqual(len({row["occurrence_id"] for row in meta}), 4) + self.assertEqual(len({row["link_id"] for row in meta}), 4) + for untouched in ('href="mailto:hello@example.test"','href="tel:+123"','href="cid:image1"','href="#local"','href="data:text/plain,hello"','href="javascript:void(0)"','href="https://postmaster.example.test/t/c/already"'): + self.assertIn(untouched, rewritten) + self.assertNotIn('href="http://one.example/a"', rewritten) + self.assertEqual(rewritten.count("https://postmaster.example.test/t/c/"), 5) + self.assertIn("HTTP One", rewritten) + self.assertIn("HTTPS Two footer", rewritten) + rows = self.links.list_links(delivery_id=self.delivery["id"], limit=20) + urls = [row["original_url"] for row in rows] + self.assertIn("https://two.example/page?a=1&b=2#section", urls) + self.assertIn("https://third.example/t/c/legitimate", urls) + repeated = [row for row in rows if row["normalized_url"] == "https://two.example/page?a=1&b=2#section"] + self.assertEqual(len(repeated), 2) + self.assertNotEqual(repeated[0]["position"], repeated[1]["position"]) + self.assertNotEqual(repeated[0]["link_id"], repeated[1]["link_id"]) + self.assertTrue(all("tracking_token" not in row for row in rows)) + + def test_click_persistence_unique_definition_filters_and_enrichment(self) -> None: + rewritten, meta = self.links.instrument_html(body_html='Project', delivery=self.delivery) + self.assertIn("/t/c/", rewritten) + link = self.links.get_by_token(self._token_for_occurrence(meta[0]["occurrence_id"])) + first = self.links.record_click(link, user_agent="Mozilla/5.0 (Windows NT 10.0) Chrome/140.0.0.0 Safari/537.36", client_ip="203.0.113.10", country_code="IT") + second = self.links.record_click(link, user_agent="Mozilla/5.0 (Windows NT 10.0) Chrome/140.0.0.0 Safari/537.36", client_ip="203.0.113.10", country_code="IT") + third = self.links.record_click(link, user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Version/18.0 Safari/605.1.15", client_ip="203.0.113.11", country_code="US") + self.assertEqual(first["event_type"], "link") + self.assertEqual(first["client_fingerprint"], second["client_fingerprint"]) + self.assertNotEqual(first["client_fingerprint"], third["client_fingerprint"]) + self.assertEqual(first["country_code"], "IT") + self.assertIn("Chrome", first["browser"]) + self.assertIn("Windows", first["os"]) + self.assertEqual(first["client_source"], "direct_or_unknown") + for kwargs in ({"campaign_id": self.campaign["id"]},{"delivery_id": self.delivery["id"]},{"link_id": meta[0]["link_id"]}): + summary = self.links.summary(**kwargs) + self.assertEqual(summary["total_clicks"], 3) + self.assertEqual(summary["unique_clicks"], 2) + self.assertEqual(summary["unique_recipients"], 1) + self.assertTrue(summary["first_click"]) + self.assertTrue(summary["last_click"]) + self.assertEqual(summary["unique_click_definition"], "delivery_id + link_id + client_fingerprint") + events = self.links.list_click_events(link_id=meta[0]["link_id"]) + self.assertEqual(len(events), 3) + self.assertEqual(events[0]["original_url"], "https://example.com/a?x=1#frag") + self.assertTrue(events[0]["user_agent"]) + top = self.links.top_links(campaign_id=self.campaign["id"]) + self.assertEqual((top[0]["total_clicks"], top[0]["unique_clicks"], top[0]["unique_recipients"]), (3,2,1)) + + def test_existing_pixel_pipeline_remains_unchanged(self) -> None: + rendered, _ = self.analytics.render_for_recipient(body_html="Hello", body_amp=None, delivery=self.delivery, track_opens=True) + self.assertIn(f"/track/open/{self.delivery['tracking_token']}.gif", rendered) + result = self.analytics.record_open(self.delivery["tracking_token"], user_agent="Mozilla/5.0 Chrome/140.0.0.0 Safari/537.36", client_ip="203.0.113.20", country_code="IT") + self.assertEqual(result["event_type"], "pixel") + self.assertEqual(self.analytics.list_open_events(delivery_id=self.delivery["id"])[0]["event_type"], "pixel") + + def test_sent_clean_placeholder_rendering_has_no_recipient_callbacks(self) -> None: + source = 'Visible labelstatus{{RECIPIENT}} {{CAMPAIGN_ID}} {{DELIVERY_ID}}' + clean = _sent_clean_html(source, self.delivery) + self.assertNotIn("/track/open/", clean) + self.assertNotIn("/t/c/", clean) + self.assertNotIn("{{TRACKING_PIXEL_URL}}", clean) + self.assertNotIn("{{AMP_STATUS_URL}}", clean) + self.assertIn("https://example.com/page?a=1&b=2#section", clean) + self.assertIn("Visible label", clean) diff --git a/tests/test_v9_4_runtime.py b/tests/test_v9_4_runtime.py new file mode 100644 index 0000000..c0267dc --- /dev/null +++ b/tests/test_v9_4_runtime.py @@ -0,0 +1,149 @@ +from __future__ import annotations + +import asyncio +import os +import tempfile +import unittest +from pathlib import Path + +from starlette.requests import Request +from starlette.routing import Mount, Route + +class RuntimeRouteTests(unittest.TestCase): + KEYS = ( + "SCHEDULER_DB_PATH", "CONTEXT_DB_PATH", "CONTEXT_SEMANTIC_ENABLED", + "CONTEXT_MODEL_AUTO_DOWNLOAD", "CONTEXT_MODEL_AUTO_PREPARE", + "FILE_STORE_DB_PATH", "FILE_STORE_ROOT", "MAIL_ACCOUNTS_DB_PATH", + "MAIL_ACCOUNTS_KEY_PATH", "EMAIL_ANALYTICS_DB_PATH", "EMAIL_ANALYTICS_KEY_PATH", + "RECIPIENT_POLICY_DB_PATH", "PUBLIC_EMAIL_BASE_URL", "PUBLIC_MCP_HOST", + "POSTMASTER_REF", "POSTMASTER_VERSION", + ) + + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + root = Path(self.tmp.name) + self.old = {key: os.environ.get(key) for key in self.KEYS} + version = (Path(__file__).resolve().parents[1] / "VERSION").read_text(encoding="utf-8").strip() + os.environ.update({ + "SCHEDULER_DB_PATH": str(root / "scheduler.db"), + "CONTEXT_DB_PATH": str(root / "knowledge.db"), + "CONTEXT_SEMANTIC_ENABLED": "false", + "CONTEXT_MODEL_AUTO_DOWNLOAD": "false", + "CONTEXT_MODEL_AUTO_PREPARE": "false", + "FILE_STORE_DB_PATH": str(root / "files.db"), + "FILE_STORE_ROOT": str(root / "files"), + "MAIL_ACCOUNTS_DB_PATH": str(root / "accounts.db"), + "MAIL_ACCOUNTS_KEY_PATH": str(root / "accounts.key"), + "EMAIL_ANALYTICS_DB_PATH": str(root / "analytics.db"), + "EMAIL_ANALYTICS_KEY_PATH": str(root / "analytics.key"), + "RECIPIENT_POLICY_DB_PATH": str(root / "policy.db"), + "PUBLIC_EMAIL_BASE_URL": "https://postmaster.example.test", + "PUBLIC_MCP_HOST": "", + "POSTMASTER_REF": f"v{version}-test", + "POSTMASTER_VERSION": "latest", + }) + from postmaster.email_analytics import analytics_store + from postmaster.link_tracking import link_store + analytics_store.cache_clear() + link_store.cache_clear() + import postmaster.runtime as runtime + self.runtime = runtime + runtime.analytics_store.cache_clear() + runtime.link_store.cache_clear() + for cached in (runtime.scheduler, runtime.context_engine, runtime.file_store, runtime.account_store, runtime.policy_client): + cached.cache_clear() + + def tearDown(self) -> None: + self.runtime.analytics_store.cache_clear() + self.runtime.link_store.cache_clear() + for cached in (self.runtime.scheduler, self.runtime.context_engine, self.runtime.file_store, self.runtime.account_store, self.runtime.policy_client): + cached.cache_clear() + for key, value in self.old.items(): + if value is None: os.environ.pop(key, None) + else: os.environ[key] = value + self.tmp.cleanup() + + @staticmethod + def _request(token: str, *, query: bytes = b"", headers: list[tuple[bytes, bytes]] | None = None) -> Request: + return Request({ + "type": "http", + "http_version": "1.1", + "method": "GET", + "scheme": "https", + "path": f"/t/c/{token}", + "raw_path": f"/t/c/{token}".encode(), + "query_string": query, + "headers": headers or [], + "client": ("203.0.113.77", 443), + "server": ("postmaster.example.test", 443), + "path_params": {"token": token}, + }) + + def test_click_route_redirects_only_to_server_record_and_records_event(self) -> None: + analytics = self.runtime.analytics_store() + links = self.runtime.link_store() + campaign = analytics.create_campaign(account_id="acct", sender="sender@example.test", subject="route", track_opens=True, amp_used=False) + delivery = analytics.create_delivery(campaign_id=campaign["id"], account_id="acct", recipient="reader@example.test", recipient_role="to") + destination = "https://destination.example/path?a=1&b=2#section" + _, meta = links.instrument_html(body_html=f'Destination', delivery=delivery) + with links._connect() as conn: + token = str(conn.execute("SELECT tracking_token FROM tracking_links WHERE id=?", (meta[0]["occurrence_id"],)).fetchone()[0]) + response = asyncio.run(self.runtime.tracking_click(self._request(token, headers=[ + (b"user-agent", b"Mozilla/5.0 Chrome/140.0.0.0 Safari/537.36"), + (b"x-forwarded-for", b"203.0.113.77"), + (b"cf-ipcountry", b"IT"), + ]))) + self.assertEqual(response.status_code, 302) + self.assertEqual(response.headers["location"], destination) + invalid = asyncio.run(self.runtime.tracking_click(self._request( + "not-a-token", query=b"url=https%3A%2F%2Fevil.example%2F" + ))) + self.assertEqual(invalid.status_code, 404) + self.assertNotIn("location", invalid.headers) + events = links.list_click_events(delivery_id=delivery["id"]) + self.assertEqual(len(events), 1) + self.assertEqual(events[0]["event_type"], "link") + self.assertEqual(events[0]["country_code"], "IT") + self.assertEqual(events[0]["original_url"], destination) + + def test_public_route_surface_build_status_dashboard_and_no_secret_exposure(self) -> None: + routes = [route for route in self.runtime.app.router.routes if isinstance(route, Route)] + paths = {route.path for route in routes} + self.assertIn("/track/open/{token}.gif", paths) + self.assertIn("/api/amp/status", paths) + self.assertIn("/t/c/{token}", paths) + self.assertIn("/files/{file_id}", paths) + self.assertEqual([p for p in paths if p.startswith("/t/c/")], ["/t/c/{token}"]) + self.assertTrue(any(isinstance(route, Mount) for route in self.runtime.app.router.routes)) + status = self.runtime.build_status() + expected_version = (Path(__file__).resolve().parents[1] / "VERSION").read_text().strip() + self.assertEqual(status["version"], expected_version) + self.assertTrue(status["native_chatgpt_file_upload"]) + self.assertTrue(status["native_file_resource_handoff"]) + self.assertTrue(status["link_tracking"]) + self.assertTrue(status["sent_copy_tracking_sanitized"]) + analytics = self.runtime.analytics_store() + links = self.runtime.link_store() + campaign = analytics.create_campaign(account_id="acct", sender="sender@example.test", subject="safe", track_opens=True, amp_used=False) + delivery = analytics.create_delivery(campaign_id=campaign["id"], account_id="acct", recipient="reader@example.test", recipient_role="to") + _, meta = links.instrument_html(body_html='Example', delivery=delivery) + with links._connect() as conn: + secret = str(conn.execute("SELECT tracking_token FROM tracking_links WHERE id=?", (meta[0]["occurrence_id"],)).fetchone()[0]) + self.assertNotIn(secret, repr(links.list_links(campaign_id=campaign["id"]))) + fragment = self.runtime._tracking_dashboard_fragment("acct") + self.assertIn("Top links", fragment) + self.assertIn("Tracking events", fragment) + self.assertNotIn(secret, fragment) + + def test_cloudflare_preflight_is_documented_not_implemented_in_yaml(self) -> None: + root = Path(__file__).resolve().parents[1] + doc = (root / "docs" / "LINK_TRACKING.md").read_text(encoding="utf-8") + self.assertIn("/t/c/*", doc) + self.assertIn("/track/open/*", doc) + self.assertIn("/api/amp/*", doc) + self.assertIn("/mcp", doc) + self.assertIn("Cloudflare Access", doc) + yaml = (root / "postmaster-mcp.yml").read_text(encoding="utf-8") + self.assertIn("POSTMASTER_VERSION", yaml) + self.assertIn("POSTMASTER_CHECK_UPDATES_ON_START", yaml) + self.assertNotIn("/t/c/*", yaml) diff --git a/tests/test_v9_4_sent_clean.py b/tests/test_v9_4_sent_clean.py new file mode 100644 index 0000000..ccf2a94 --- /dev/null +++ b/tests/test_v9_4_sent_clean.py @@ -0,0 +1,111 @@ +from __future__ import annotations + +import base64 +import inspect +import os +import re +import tempfile +import unittest +from email.message import EmailMessage +from pathlib import Path +from unittest.mock import patch + +from postmaster.email_analytics import EmailAnalyticsStore +from postmaster.link_tracking import LinkTrackingStore +from postmaster.link_tracking_html import eligible_web_url +from postmaster.tracked_mail import LinkTrackingMailClient, _sent_clean_html, _synchronize_transport_headers +from postmaster.mail_bridge import MailClient, Settings + +class CapturingV94MailClient(LinkTrackingMailClient): + def __init__(self, settings: Settings): + super().__init__(settings) + self.outbound: list[EmailMessage] = [] + self.sent: list[EmailMessage] = [] + + def _validate_recipients(self, recipients): + return [str(x).strip() for x in recipients if str(x).strip()] + + def _send_message_with_clean_sent(self, outbound, sent_copy, recipients): + _synchronize_transport_headers(outbound, sent_copy, self.settings.email_address) + self.outbound.append(outbound) + self.sent.append(sent_copy) + return {"sent": True,"from": self.settings.email_address,"to": recipients,"subject": str(outbound.get("Subject", "")),"message_id": str(outbound.get("Message-ID", "")),"sent_copy_saved": True,"sent_copy_error": None,"sent_copy_tracking_sanitized": True} + + +class SentVariantTests(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + root = Path(self.tmp.name) + self.old_public = {k: os.environ.get(k) for k in ("PUBLIC_EMAIL_BASE_URL", "PUBLIC_MCP_HOST")} + os.environ["PUBLIC_EMAIL_BASE_URL"] = "https://postmaster.example.test" + os.environ["PUBLIC_MCP_HOST"] = "" + self.analytics = EmailAnalyticsStore(db_path=str(root / "analytics.db"), key_path=str(root / "analytics.key")) + self.links = LinkTrackingStore(self.analytics) + self.settings = Settings(email_address="sender@example.test", email_password="pw", enable_send=True, save_sent_copy=True, send_recipient_allowlist=("example.test",), allow_previous_sent_recipients=False, account_id="acct", smtp_username="sender@example.test", smtp_password="pw") + self.client = CapturingV94MailClient(self.settings) + + def tearDown(self) -> None: + for key, value in self.old_public.items(): + if value is None: os.environ.pop(key, None) + else: os.environ[key] = value + self.tmp.cleanup() + + @staticmethod + def _part_text(msg: EmailMessage, content_type: str) -> str: + for part in msg.walk(): + if part.get_content_type() == content_type: + return str(part.get_content()) + return "" + + @staticmethod + def _attachment_bytes(msg: EmailMessage) -> list[bytes]: + return [part.get_payload(decode=True) or b"" for part in msg.walk() if part.get_content_disposition() == "attachment"] + + def test_pre_v94_sent_behavior_is_same_instrumented_message(self) -> None: + source = inspect.getsource(MailClient._send_message) + self.assertIn("smtp.send_message(msg", source) + self.assertIn("msg.as_bytes(policy=policy.SMTP)", source) + self.assertIn("conn.append", source) + + def test_recipient_is_tracked_but_sent_is_clean_with_headers_and_attachment_identity(self) -> None: + payload = b"\x00v9.4-attachment-bytes\xff" + body_html = 'OneTwo' + attachments = [{"filename":"asset.bin","content_type":"application/octet-stream","content_base64":base64.b64encode(payload).decode("ascii")}] + with patch("postmaster.tracked_mail.analytics_store", return_value=self.analytics), patch("postmaster.tracked_mail.link_store", return_value=self.links): + result = self.client._send_individualized(to=["reader@example.test"], subject="Tracked test", body="Plain fallback", body_html=body_html, attachments=attachments, track_opens=True, in_reply_to="", references=" ") + self.assertTrue(result["sent"]) + self.assertTrue(result["sent_copy_tracking_sanitized"]) + outbound, sent = self.client.outbound[0], self.client.sent[0] + outbound_html = self._part_text(outbound, "text/html") + sent_html = self._part_text(sent, "text/html") + self.assertIn("/track/open/", outbound_html) + self.assertGreaterEqual(outbound_html.count("/t/c/"), 2) + self.assertNotIn("/track/open/", sent_html) + self.assertNotIn("/t/c/", sent_html) + self.assertIn("https://one.example/a?x=1&y=2#frag", sent_html) + self.assertIn("https://two.example/project", sent_html) + self.assertIn("One", sent_html) + self.assertIn("Two", sent_html) + for header in ("Message-ID","Date","Subject","In-Reply-To","References"): + self.assertEqual(str(outbound.get(header, "")), str(sent.get(header, "")), header) + self.assertEqual(self._attachment_bytes(outbound), [payload]) + self.assertEqual(self._attachment_bytes(sent), [payload]) + delivery_id = result["deliveries"][0]["delivery_id"] + delivery = self.analytics.get_delivery(delivery_id) + self.assertEqual(delivery["campaign_id"], result["campaign_id"]) + self.assertEqual(delivery["message_id"], str(outbound["Message-ID"])) + links = self.links.list_links(delivery_id=delivery_id) + self.assertEqual(len(links), 2) + self.assertTrue(all(row["message_id"] == str(outbound["Message-ID"]) for row in links)) + sent_hrefs = re.findall(r'href="([^"]+)"', sent_html) + self.assertTrue(sent_hrefs) + self.assertTrue(all("/t/c/" not in href for href in sent_hrefs)) + self.assertTrue(all(eligible_web_url(href.replace("&", "&")) for href in sent_hrefs)) + self.assertEqual(self.links.summary(delivery_id=delivery_id)["total_clicks"], 0) + self.assertEqual(len(self.analytics.list_open_events(delivery_id=delivery_id)), 0) + + def test_old_messages_are_not_retroactively_mutated(self) -> None: + legacy = b'old' + before = bytes(legacy) + _ = _sent_clean_html("

new

", {"recipient":"r@example.test","campaign_id":"c","id":"d"}) + self.assertEqual(legacy, before)