feat(messaging): give transactional email its own scope

frontend/src/lib/oauthScopes.generated.ts

@@ -214,6 +214,8 @@ export const OAUTH_SCOPES_SUPPORTED = [
     'toolbar:write',
     'tracing:read',
     'tracing:write',
+    'transactional_email:read',
+    'transactional_email:write',
     'field_note:read',
     'field_note:write',
     'uploaded_media:read',
@@ -268,4 +270,5 @@ export const OAUTH_SCOPES_HIDDEN = [
 export const LEGACY_SCOPE_FALLBACKS: ReadonlyMap<string, string> = new Map([
     ['inbox', 'task'],
     ['messaging_preference', 'hog_flow'],
+    ['transactional_email', 'hog_flow'],
 ])

frontend/src/lib/scopes.tsx

@@ -312,6 +312,12 @@ export const API_SCOPES: APIScope[] = [
     { key: 'ticket', objectName: 'Ticket', objectPlural: 'tickets' },
     { key: 'toolbar', objectName: 'Toolbar', objectPlural: 'toolbar' },
     { key: 'tracing', objectName: 'Tracing', objectPlural: 'tracing' },
+    {
+        key: 'transactional_email',
+        objectName: 'Transactional email',
+        objectPlural: 'transactional emails',
+        info: "Write sends transactional email from a template. Read gets a sent email's status and works only with a project secret API key. This scope can't change templates or workflows.",
+    },
     { key: 'field_note', objectName: 'Field note', objectPlural: 'field notes' },
     { key: 'uploaded_media', objectName: 'Uploaded media', objectPlural: 'uploaded media' },
     { key: 'usage_metric', objectName: 'Usage metric', objectPlural: 'usage metrics' },
@@ -391,6 +397,8 @@ export const PROJECT_SECRET_API_KEY_ALLOWED_API_SCOPE_ACTION = [
     'hog_flow:read',
     'hog_flow:write',
     'messaging_preference:write',
+    'transactional_email:read',
+    'transactional_email:write',
 ] as const
 
 export type ProjectSecretAPIKeyAllowedScope = (typeof PROJECT_SECRET_API_KEY_ALLOWED_API_SCOPE_ACTION)[number]
@@ -506,6 +514,7 @@ export const PROJECT_SECRET_API_KEY_SCOPE_PRESETS: ProjectSecretAPIKeyScopePrese
     { value: 'local_evaluation', label: 'Local feature flag evaluation', scopes: ['feature_flag:read'] },
     { value: 'llm_gateway', label: 'AI gateway access', scopes: ['llm_gateway:read'] },
     { value: 'recipient_preferences', label: 'Set recipient preferences', scopes: ['messaging_preference:write'] },
+    { value: 'transactional_email', label: 'Send transactional email', scopes: ['transactional_email:write'] },
 ]
 
 // The product areas that the scope pickers use to group objects, in display order. Each scope object
@@ -612,6 +621,10 @@ export const API_SCOPE_GROUPS: APIScopeGroup[] = [
             'plugin',
         ],
     },
+    {
+        label: 'Messaging',
+        objects: ['transactional_email'],
+    },
     {
         label: 'AI & context management tools',
         objects: [

posthog/scopes.py

@@ -138,6 +138,7 @@
     "today",
     "toolbar",
     "tracing",
+    "transactional_email",
     "field_note",
     "uploaded_media",
     "usage_metric",
@@ -285,6 +286,8 @@
     ("hog_flow", "read"),
     ("hog_flow", "write"),
     ("messaging_preference", "write"),
+    ("transactional_email", "read"),
+    ("transactional_email", "write"),
 ]
 
 # Server-side scope assignment string-set constants (see RFC: server-side scope
@@ -496,6 +499,8 @@ def scopes_not_covered(held_scopes: Iterable[str], required_scopes: Iterable[str
     "inbox": "task",
     # Permanent: `hog_flow:write` stays a supported way to set recipient preferences. Keep it on low use.
     "messaging_preference": "hog_flow",
+    # Permanent: `hog_flow` stays a supported way to send transactional email and read a send's status.
+    "transactional_email": "hog_flow",
 }
 
 

products/messaging/backend/presentation/views/message_templates.py

@@ -34,7 +34,7 @@
 from posthog.integrations import IntegrationManager, IntegrationNotFound
 from posthog.models import User
 from posthog.models.tag import tagify
-from posthog.permissions import is_authenticated_via_project_secret_api_key
+from posthog.permissions import is_authenticated_via_project_secret_api_key, legacy_scope_object_used
 from posthog.rate_limit import PersonalOrProjectSecretApiKeyRateThrottle, ProjectSecretApiKeyTeamRateThrottle
 
 from products.cdp.backend.presentation.hog_functions import build_html_wrap_design
@@ -513,6 +513,15 @@ class TransactionalEmailTeamSustainedThrottle(ProjectSecretApiKeyTeamRateThrottl
     rate = "20000/hour"
 
 
+# LEGACY_SCOPE_FALLBACKS lets a `hog_flow` key satisfy these scopes too.
+TRANSACTIONAL_EMAIL_SEND_SCOPE = "transactional_email:write"
+TRANSACTIONAL_EMAIL_STATUS_SCOPE = "transactional_email:read"
+
+
+def transactional_email_authorized_by(request: Any, required_scope: str) -> str:
+    return legacy_scope_object_used(request, required_scope) or "transactional_email"
+
+
 def require_transactional_email_api(team: Any) -> None:
     if not posthoganalytics.feature_enabled(
         TRANSACTIONAL_EMAIL_API_FLAG,
@@ -674,6 +683,7 @@ def _report_transactional_email_send(
                     "auth_method": "project_secret_api_key"
                     if is_authenticated_via_project_secret_api_key(request)
                     else "user",
+                    "authorized_by": transactional_email_authorized_by(request, TRANSACTIONAL_EMAIL_SEND_SCOPE),
                     **outcome,
                 },
                 groups=groups(self.organization, self.team),
@@ -696,16 +706,15 @@ class MessageTemplatesViewSet(
 ):
     scope_object = "hog_flow"
     permission_classes = [IsAuthenticated]
-    # `design` and `send` are custom write actions; list them so programmatic callers (MCP/personal API key)
-    # get hog_flow:write checked instead of being rejected as an action with no declared scope.
+    # `design` and `make_transactional` are custom write actions; list them so programmatic callers (MCP/personal
+    # API key) get hog_flow:write checked instead of being rejected as an action with no declared scope.
     scope_object_write_actions = [
         "create",
         "update",
         "partial_update",
         "patch",
         "destroy",
         "design",
-        "send",
         "make_transactional",
     ]
     authentication_classes = [ProjectSecretAPIKeyAuthentication]
@@ -912,11 +921,12 @@ def apply_operations(content: dict[str, Any]) -> dict[str, Any]:
             "Send one transactional email from this template to one recipient. The template must be made "
             "transactional first. `message_data` fills the template's `{{ variables.* }}` placeholders. "
             "Transactional emails skip opt-outs and frequency caps. Send an `Idempotency-Key` header to retry "
-            "safely without sending the email twice."
+            "safely without sending the email twice. Needs the `transactional_email:write` scope. A key with "
+            "`hog_flow:write` also works, but that scope reaches the rest of Workflows too."
         ),
         parameters=[TEMPLATE_ID_PARAMETER, IDEMPOTENCY_KEY_PARAMETER],
     )
-    @action(detail=True, methods=["POST"], pagination_class=None)
+    @action(detail=True, methods=["POST"], pagination_class=None, required_scopes=[TRANSACTIONAL_EMAIL_SEND_SCOPE])
     def send(self, request: Any, *args: Any, **kwargs: Any) -> Response:
         self._end_request_validation()
         with transactional_email_stage("feature_flag"):
@@ -978,7 +988,6 @@ def make_transactional(self, request: Any, *args: Any, **kwargs: Any) -> Respons
 
 class MessagingSendViewSet(TransactionalEmailSendMixin, TeamAndOrgViewSetMixin, viewsets.ViewSet):
     scope_object = "hog_flow"
-    scope_object_write_actions = ["email"]
     permission_classes = [IsAuthenticated]
     authentication_classes = [ProjectSecretAPIKeyAuthentication]
     psak_allowed_actions = ["email"]
@@ -995,11 +1004,12 @@ def get_throttles(self):
             "Send one transactional email to one recipient from the template named in `template_id`. The template "
             "must be made transactional first. `message_data` fills the template's `{{ variables.* }}` placeholders. "
             "Transactional emails skip opt-outs and frequency caps. Send an `Idempotency-Key` header to retry "
-            "safely without sending the email twice."
+            "safely without sending the email twice. Needs the `transactional_email:write` scope. A key with "
+            "`hog_flow:write` also works, but that scope reaches the rest of Workflows too."
         ),
         parameters=[IDEMPOTENCY_KEY_PARAMETER],
     )
-    @action(detail=False, methods=["POST"])
+    @action(detail=False, methods=["POST"], required_scopes=[TRANSACTIONAL_EMAIL_SEND_SCOPE])
     def email(self, request: Any, *args: Any, **kwargs: Any) -> Response:
         self._end_request_validation()
         with transactional_email_stage("feature_flag"):

products/messaging/backend/presentation/views/transactional_email_status.py

@@ -19,7 +19,11 @@
 from posthog.rate_limit import PersonalOrProjectSecretApiKeyRateThrottle, ProjectSecretApiKeyTeamRateThrottle
 
 from products.messaging.backend.facade.transactional_email import EmailDeliveryStatus, EmailStatus, get_email_status
-from products.messaging.backend.presentation.views.message_templates import require_transactional_email_api
+from products.messaging.backend.presentation.views.message_templates import (
+    TRANSACTIONAL_EMAIL_STATUS_SCOPE,
+    require_transactional_email_api,
+    transactional_email_authorized_by,
+)
 
 logger = structlog.get_logger(__name__)
 
@@ -74,7 +78,7 @@ class TransactionalEmailReadTeamSustainedThrottle(ProjectSecretApiKeyTeamRateThr
 
 class MessagingEmailsViewSet(TeamAndOrgViewSetMixin, viewsets.ViewSet):
     scope_object = "hog_flow"
-    scope_object_read_actions = ["retrieve"]
+    required_scopes = [TRANSACTIONAL_EMAIL_STATUS_SCOPE]
     permission_classes = [IsAuthenticated]
     authentication_classes = [ProjectSecretAPIKeyAuthentication]
     psak_allowed_actions = ["retrieve"]
@@ -104,7 +108,8 @@ def get_throttles(self) -> list[BaseThrottle]:
         summary="Get a transactional email's status",
         description=(
             "Get the status of one transactional email by the `message_id` that the send returned, for 30 days after "
-            "the send. Needs a project secret API key with the `hog_flow:read` scope."
+            "the send. Needs a project secret API key with the `transactional_email:read` scope. A key with "
+            "`hog_flow:read` also works, but that scope reaches the rest of Workflows too."
         ),
     )
     def retrieve(self, request: Request, *args: Any, pk: str | None = None, **kwargs: Any) -> Response:
@@ -140,6 +145,8 @@ def _report_status_read(self, request: Request, email_status: EmailStatus | None
                     "team_id": self.team_id,
                     "outcome": "not_found" if email_status is None else "found",
                     "status": email_status.status if email_status else None,
+                    "auth_method": "project_secret_api_key",
+                    "authorized_by": transactional_email_authorized_by(request, TRANSACTIONAL_EMAIL_STATUS_SCOPE),
                 },
                 groups=groups(self.organization, self.team),
             )

products/messaging/backend/tests/api/test_transactional_email.py

@@ -161,6 +161,8 @@ def test_passes_the_idempotency_key_to_the_pipeline_and_reports_a_replay(self, m
             ("send_with_write_scope", "post", "send/", ["hog_flow:write"], status.HTTP_202_ACCEPTED),
             ("send_with_read_scope", "post", "send/", ["hog_flow:read"], status.HTTP_403_FORBIDDEN),
             ("send_with_preference_scope", "post", "send/", ["messaging_preference:write"], status.HTTP_403_FORBIDDEN),
+            ("send_with_send_scope", "post", "send/", ["transactional_email:write"], status.HTTP_202_ACCEPTED),
+            ("send_with_status_scope", "post", "send/", ["transactional_email:read"], status.HTTP_403_FORBIDDEN),
             ("send_email_with_write_scope", "post", "send_email", ["hog_flow:write"], status.HTTP_202_ACCEPTED),
             ("send_email_with_read_scope", "post", "send_email", ["hog_flow:read"], status.HTTP_403_FORBIDDEN),
             (
@@ -170,6 +172,20 @@ def test_passes_the_idempotency_key_to_the_pipeline_and_reports_a_replay(self, m
                 ["messaging_preference:write"],
                 status.HTTP_403_FORBIDDEN,
             ),
+            (
+                "send_email_with_send_scope",
+                "post",
+                "send_email",
+                ["transactional_email:write"],
+                status.HTTP_202_ACCEPTED,
+            ),
+            (
+                "send_email_with_status_scope",
+                "post",
+                "send_email",
+                ["transactional_email:read"],
+                status.HTTP_403_FORBIDDEN,
+            ),
             ("retrieve_is_not_a_psak_action", "get", "", ["hog_flow:write"], status.HTTP_403_FORBIDDEN),
             (
                 "make_transactional_is_not_a_psak_action",
@@ -193,6 +209,49 @@ def test_project_secret_api_key_can_only_send(
 
         assert response.status_code == expected_status, response.json()
 
+    @parameterized.expand(
+        [
+            ("send_scope", "template", ["transactional_email:write"], "transactional_email"),
+            ("workflow_scope", "template", ["hog_flow:write"], "hog_flow"),
+            (
+                "send_scope_on_send_email",
+                "send_email",
+                ["transactional_email:write"],
+                "transactional_email",
+            ),
+            ("workflow_scope_on_send_email", "send_email", ["hog_flow:write"], "hog_flow"),
+            (
+                "both_scopes",
+                "template",
+                ["hog_flow:write", "transactional_email:write"],
+                "transactional_email",
+            ),
+        ]
+    )
+    def test_records_the_scope_that_authorized_the_send(
+        self, mock_post, _flag, capture, _name, route, scopes, expected_authorized_by
+    ):
+        self._cdp_accepts(mock_post)
+
+        response = self._send(self._psak_client(scopes), route, self.body)
+
+        assert response.status_code == status.HTTP_202_ACCEPTED, response.json()
+        assert capture.call_args.kwargs["properties"]["authorized_by"] == expected_authorized_by
+
+    def test_a_personal_api_key_with_the_send_scope_cannot_change_the_template(self, mock_post, _flag, _capture):
+        self._cdp_accepts(mock_post)
+        client = APIClient()
+        client.credentials(
+            HTTP_AUTHORIZATION=f"Bearer {self.create_personal_api_key_with_scopes(['transactional_email:write'])}"
+        )
+        template_url = f"/api/projects/{self.team.id}/messaging_templates/{self.template.id}/"
+
+        send = client.post(self.url, self.body, format="json")
+        edit = client.patch(template_url, {"name": "Renamed"}, format="json")
+
+        assert send.status_code == status.HTTP_202_ACCEPTED, send.json()
+        assert edit.status_code == status.HTTP_403_FORBIDDEN, edit.json()
+
     @parameterized.expand(
         [
             ("not_transactional", "template_not_transactional"),

products/messaging/backend/tests/api/test_transactional_email_status.py

@@ -122,6 +122,8 @@ def test_reports_the_latest_status_of_the_send(self, _flag, _name, rows, expecte
             "team_id": self.team.id,
             "outcome": "found",
             "status": expected_status,
+            "auth_method": "project_secret_api_key",
+            "authorized_by": "hog_flow",
         }
         assert response.json() == {
             "id": self.message_id,
@@ -155,17 +157,26 @@ def test_answers_404_for_an_email_it_cannot_report(self, _flag, _name, message_i
 
     @parameterized.expand(
         [
-            ("read_scope", ["hog_flow:read"], status.HTTP_200_OK),
-            ("no_workflow_scope", ["feature_flag:read"], status.HTTP_403_FORBIDDEN),
-            ("preference_scope", ["messaging_preference:write"], status.HTTP_403_FORBIDDEN),
+            ("read_scope", ["hog_flow:read"], status.HTTP_200_OK, "hog_flow"),
+            ("status_scope", ["transactional_email:read"], status.HTTP_200_OK, "transactional_email"),
+            ("send_scope", ["transactional_email:write"], status.HTTP_200_OK, "transactional_email"),
+            ("no_workflow_scope", ["feature_flag:read"], status.HTTP_403_FORBIDDEN, None),
+            ("preference_scope", ["messaging_preference:write"], status.HTTP_403_FORBIDDEN, None),
         ]
     )
-    def test_project_secret_api_key_needs_the_read_scope(self, _flag, _name, scopes, expected_status):
+    def test_project_secret_api_key_needs_the_read_scope(
+        self, _flag, _name, scopes, expected_status, expected_authorized_by
+    ):
         self._status_rows([("queued", 0, "")])
 
-        response = self._psak_client(scopes).get(self._url(self.message_id))
+        with patch(
+            "products.messaging.backend.presentation.views.transactional_email_status.posthoganalytics.capture"
+        ) as capture:
+            response = self._psak_client(scopes).get(self._url(self.message_id))
 
         assert response.status_code == expected_status
+        if expected_authorized_by:
+            assert capture.call_args.kwargs["properties"]["authorized_by"] == expected_authorized_by
 
     def test_a_session_cannot_read_a_status(self, _flag):
         self._status_rows([("queued", 0, "")])