--- /dev/null +++ b/src/user_engine/mail_bindings.py @@ -0,0 +1,40 @@ +"""Explicit tenant globs for the scoped invitation mail transport.""" + +from __future__ import annotations + +import json +import re +from fnmatch import fnmatchcase + +IDENTIFIER = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}\Z") +PATTERN = re.compile(r"[A-Za-z0-9*?][A-Za-z0-9._:*?-]{0,127}\Z") + + +def load_mail_client_bindings(raw): + if raw is None: + return None + try: + rows = json.loads(raw) + if not isinstance(rows, list) or not 1 <= len(rows) <= 128: + raise ValueError + result = [] + for row in rows: + if (not isinstance(row, list) or len(row) != 2 + or not isinstance(row[0], str) or not PATTERN.fullmatch(row[0]) + or not isinstance(row[1], str) or not IDENTIFIER.fullmatch(row[1])): + raise ValueError + result.append(tuple(row)) + if len({row[0] for row in result}) != len(result): + raise ValueError + return tuple(result) + except (ValueError, TypeError): + raise ValueError("mail_client_bindings_invalid") from None + + +def mail_client_for(tenant, bindings): + if not isinstance(tenant, str) or not IDENTIFIER.fullmatch(tenant): + raise ValueError("mail_binding_denied") + matches = {client for pattern, client in bindings if fnmatchcase(tenant, pattern)} + if len(matches) != 1: + raise ValueError("mail_binding_denied") + return matches.pop() --- a/src/user_engine/adapters/delivery.py +++ b/src/user_engine/adapters/delivery.py @@ -19,6 +19,7 @@ from user_engine.domain import OutboxEvent +from user_engine.mail_bindings import load_mail_client_bindings, mail_client_for _MAIL_EVENTS = {"family_member.invited", "family_invitation.resent"} @@ -34,12 +35,14 @@ mail_url: str | None = None, mail_bearer_token: str | None = None, timeout_seconds: float = 5.0, + mail_client_bindings: str | None = None, ) -> None: self.event_url = event_url self.mail_url = mail_url self.event_bearer_token = event_bearer_token self.mail_bearer_token = mail_bearer_token self.timeout_seconds = timeout_seconds + self.mail_client_bindings = load_mail_client_bindings(mail_client_bindings) def __call__(self, event: OutboxEvent) -> None: envelope = { @@ -57,7 +60,15 @@ if self.mail_url and event.event_type in _MAIL_EVENTS: if not self.mail_bearer_token: raise RuntimeError("mail delivery token is required") - self._post(self.mail_url, envelope, self.mail_bearer_token) + mail_envelope = dict(envelope) + if self.mail_client_bindings is not None: + try: + mail_envelope["client_id"] = mail_client_for( + event.tenant, self.mail_client_bindings + ) + except ValueError: + raise DeliveryError("mail_binding_denied") from None + self._post(self.mail_url, mail_envelope, self.mail_bearer_token) event_envelope = dict(envelope) event_data = dict(envelope["data"]) if "primary_email" in event_data: --- a/src/user_engine/runtime.py +++ b/src/user_engine/runtime.py @@ -65,6 +65,7 @@ else None ), timeout_seconds=float(os.environ.get("USER_ENGINE_DELIVERY_TIMEOUT", "5")), + mail_client_bindings=os.environ.get("USER_ENGINE_MAIL_CLIENT_BINDINGS"), ) app = PortalApplication( service, --- /dev/null +++ b/tests/test_mail_bindings.py @@ -0,0 +1,73 @@ +"""Scoped mail migration uses synthetic identities and a mocked HTTP boundary.""" + +import json +import unittest +from unittest.mock import patch + +from user_engine.adapters.delivery import HTTPOutboxDeliveryAdapter, DeliveryError +from user_engine.domain import OutboxEvent +from user_engine.mail_bindings import load_mail_client_bindings, mail_client_for + + +class MailBindingsTests(unittest.TestCase): + def test_whole_case_sensitive_globs_and_literal_clients(self): + bindings = load_mail_client_bindings('[ ["tenant:*-family", "coulomb-social"] ]') + self.assertEqual(mail_client_for("tenant:synthetic-family", bindings), "coulomb-social") + for tenant in ("tenant:platform", "tenant:Synthetic-Family", "tenant:x-family/path"): + with self.assertRaisesRegex(ValueError, "mail_binding_denied"): + mail_client_for(tenant, bindings) + + def test_invalid_configuration_never_falls_back(self): + for raw in ('', '[]', '{}', '[["*", "*"]]', '[["[a]", "client"]]', + '[["*", "one"], ["*", "two"]]', '[[true, "client"]]'): + with self.assertRaisesRegex(ValueError, "mail_client_bindings_invalid"): + load_mail_client_bindings(raw) + self.assertIsNone(load_mail_client_bindings(None)) + + def test_conflicting_overlaps_are_denied(self): + bindings = load_mail_client_bindings('[["tenant:*", "one"], ["tenant:x", "two"]]') + with self.assertRaisesRegex(ValueError, "mail_binding_denied"): + mail_client_for("tenant:x", bindings) + + def adapter(self, bindings=None): + return HTTPOutboxDeliveryAdapter(event_url="http://events.example.test", + event_bearer_token="synthetic-event-token", mail_url="http://mail.example.test", + mail_bearer_token="synthetic-mail-token", mail_client_bindings=bindings) + + def event(self, kind="family_member.invited", tenant="tenant:synthetic-family"): + return OutboxEvent(event_id="evt-synthetic", event_type=kind, aggregate_id="inv-synthetic", + payload={"primary_email": "member@example.test", "invitation_id": "inv-synthetic"}, + tenant=tenant, correlation_id="trace-synthetic") + + def test_binding_only_changes_mail_transport_not_event_or_outbox(self): + event = self.event() + adapter = self.adapter('[["tenant:*-family", "coulomb-social"]]') + with patch.object(adapter, "_post") as post: + adapter(event) + mail, audit = [call.args[1] for call in post.call_args_list] + self.assertEqual(mail["client_id"], "coulomb-social") + self.assertNotIn("client_id", audit) + self.assertNotIn("primary_email", audit["data"]) + self.assertNotIn("client_id", event.payload) + self.assertEqual(mail["id"], event.event_id) + + def test_legacy_mode_and_non_mail_events_preserve_envelopes(self): + adapter = self.adapter() + with patch.object(adapter, "_post") as post: + adapter(self.event()) + self.assertNotIn("client_id", post.call_args_list[0].args[1]) + adapter = self.adapter('[["tenant:*-family", "coulomb-social"]]') + with patch.object(adapter, "_post") as post: + adapter(self.event("membership.added", "tenant:platform")) + self.assertEqual(post.call_count, 1) + + def test_unadmitted_mail_tenant_contacts_neither_lane(self): + adapter = self.adapter('[["tenant:*-family", "coulomb-social"]]') + with patch.object(adapter, "_post") as post: + with self.assertRaisesRegex(DeliveryError, "mail_binding_denied"): + adapter(self.event(tenant="tenant:platform")) + post.assert_not_called() + + +if __name__ == "__main__": + unittest.main()