Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .github/scripts/check-idor-model-coverage.py
Original file line number Diff line number Diff line change
Expand Up @@ -329,6 +329,7 @@ def get_scoped_models() -> tuple[dict[str, set[str]], set[str], set[str], set[st
"DevStackImageTarget",
"DevStackImageAttempt",
"License",
"SandboxSenderTenantState", # instance-global mirror of the shared sandbox SES tenant state
# --- Deprecated ---
"Prompt",
"PromptSequence",
Expand Down Expand Up @@ -361,6 +362,7 @@ def get_scoped_models() -> tuple[dict[str, set[str]], set[str], set[str], set[st
"RoleMembership",
"LinkedIdentityProviderConfig",
"IdJagIdentity",
"SandboxSenderOrganizationBlock",
# --- User-scoped (cross-tenant by design) ---
"NotificationViewed",
"SCIMProvisionedUser",
Expand Down
2 changes: 2 additions & 0 deletions .semgrep/rules/security/idor-team-scoped-models.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -1407,6 +1407,7 @@ rules:
|Role
|RoleExternalReference
|RoleMembership
|SandboxSenderOrganizationBlock
)$
- focus-metavariable: $SINK
pattern-sanitizers:
Expand Down Expand Up @@ -1490,6 +1491,7 @@ rules:
|Role
|RoleExternalReference
|RoleMembership
|SandboxSenderOrganizationBlock
)$
- pattern-not: $M.objects. ... .$METHOD2(..., organization=$O, ...)
- pattern-not: $M.objects. ... .$METHOD2(..., organization_id=$O, ...)
Expand Down
7 changes: 7 additions & 0 deletions docs/internal/workflows-email-sending-tiers.md
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,8 @@ Only workflows that send email are subject to the tiers; SMS, push, and webhook

The `workflows-sandbox-sender` flag enables a PostHog-managed email integration per project through `POST /api/projects/:project_id/integrations/email_sandbox_sender/`.
The endpoint requires `integration:write` and project membership. Repeated requests keep the same integration and refresh its display name from the organization name.
Staff turn the sandbox sender off for one organization by adding a `SandboxSenderOrganizationBlock` row in the Django admin, and lift the block by deleting the row.
While the block exists, the endpoint returns 404 for the organization's projects. Existing sandbox senders stay.
Set `WORKFLOWS_SANDBOX_SENDER_DOMAIN`, `SES_SANDBOX_FROM_ADDRESS`, and `SES_SANDBOX_TENANT_NAME` in Django; any empty value leaves the endpoint unavailable and lets every team register any domain as before.
The worker sends from the same `SES_SANDBOX_FROM_ADDRESS`, so set it once for Django and the worker. The address the UI shows is then the address the worker sends from.
With the flag on, opening the Channels page creates the sender and shows it in its own row.
Expand Down Expand Up @@ -111,5 +113,10 @@ A reached allowance skips the send instead of delaying it. A limiter error, or a
The worker checks members after the claim and gives the claim back when that check skips the send.
An SES throttle response also gives the claim back before the delivery retry. Returns are deduplicated, and Valkey ignores a return that arrives more than five minutes after it was issued.
Test sends run inline in the CDP API, so a sandbox-enabled CDP API opens the SES Valkey pool (`SES_RATE_LIMITER_VALKEY_*`) for the allowances. Without it, every sandbox test send skips.
Django mirrors the tenant named by `SES_SANDBOX_TENANT_NAME` into `SandboxSenderTenantState` from the SES tenant-state webhook and the daily reconciliation, without emailing project admins. Set the same `SES_SANDBOX_TENANT_NAME` on Django web, Celery, and the worker.
A committed state change reaches the workers through the `reload-sandbox-sender-state` message.
Each delivered send emits a `workflows sandbox email sent` event.
Live workflow sends record `email_sent`, so workflow reports count them, and also `email_sandbox_sent`, so the sending tier and the sending allowance can leave them out; test sends record neither.
The sending tier leaves `email_sandbox_sent` out of its volume and its rate denominators, and the sending allowance leaves it out too, floored at zero per day and per workflow.
While `WORKFLOWS_SANDBOX_SENDER_DOMAIN`, `SES_SANDBOX_FROM_ADDRESS`, or `SES_SANDBOX_TENANT_NAME` is empty, both read only `email_sent`, as before.
Bounces and complaints of sandbox sends still count, so the exclusion can only raise a rate.
5 changes: 5 additions & 0 deletions posthog/plugins/plugin_server_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,11 @@ def reload_team_workflows_config_on_workers(team_id: int):
publish_message("reload-team-workflows-config", {"teamId": team_id})


def reload_sandbox_sender_state_on_workers(tenant_name: str) -> None:
logger.info(f"Reloading sandbox sender state for tenant {tenant_name} on workers")
publish_message("reload-sandbox-sender-state", {"tenantName": tenant_name})

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 · Wire the sandbox reload message to a worker consumer

When reconciliation records DISABLED, this publishes to a channel that no worker subscribes to. Workers also never read SandboxSenderTenantState; SandboxEmailSender.canSend depends only on configuration, so they continue attempting sends against the paused tenant. Add a worker state reader and subscriber, and enforce that state before sending.



def populate_plugin_capabilities_on_workers(plugin_id: str):
logger.info(f"Populating plugin capabilities for plugin {plugin_id} on workers")
publish_message("populate-plugin-capabilities", {"pluginId": plugin_id})
Expand Down
1 change: 1 addition & 0 deletions products/workflows/backend/admin/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,5 +4,6 @@
hog_flow_admin,
hog_flow_batch_job_admin,
hog_flow_template_admin,
sandbox_sender_organization_block_admin,
team_workflows_config_admin,
)
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
from typing import cast

from django.contrib import admin
from django.forms import ModelForm
from django.http import HttpRequest

from posthog.models.user import User

from products.workflows.backend.models import SandboxSenderOrganizationBlock


@admin.register(SandboxSenderOrganizationBlock)
class SandboxSenderOrganizationBlockAdmin(admin.ModelAdmin):
list_display = ("organization", "reason", "created_by", "created_at")
fields = ("organization", "reason", "created_by", "created_at")
readonly_fields = ("created_by", "created_at")
raw_id_fields = ("organization",)
search_fields = ("organization__id", "organization__name", "reason")
list_select_related = ("organization", "created_by")
ordering = ("-created_at",)

def save_model(
self,
request: HttpRequest,
obj: SandboxSenderOrganizationBlock,
form: ModelForm | None,
change: bool,
) -> None:
if not change:
obj.created_by = cast(User, request.user)
super().save_model(request, obj, form, change)
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
# Generated by Django 6.1.2 on 2026-11-03 14:30

from django.db import migrations, models


class Migration(migrations.Migration):
dependencies = [
("workflows", "0040_team_workflows_config_frequency_cap_hold"),
]

operations = [
migrations.CreateModel(
name="SandboxSenderTenantState",
fields=[
(
"id",
models.BigAutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
("tenant_name", models.CharField(max_length=64, unique=True)),
(
"sending_status",
models.CharField(blank=True, db_default="", default="", max_length=32),
),
("synced_at", models.DateTimeField(blank=True, null=True)),
],
),
]
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
# Generated by Django 6.1.2 on 2026-11-03 14:30

import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models


class Migration(migrations.Migration):
dependencies = [
("posthog", "1411_personalapikey_expires_at"),
("workflows", "0041_sandbox_sender_tenant_state"),
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]

operations = [
migrations.CreateModel(
name="SandboxSenderOrganizationBlock",
fields=[
(
"id",
models.BigAutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
("reason", models.TextField()),
("created_at", models.DateTimeField(auto_now_add=True)),
(
"created_by",
models.ForeignKey(
blank=True,
db_constraint=False,
null=True,
on_delete=django.db.models.deletion.SET_NULL,
related_name="+",
to=settings.AUTH_USER_MODEL,
),
),
(
"organization",
models.OneToOneField(
db_constraint=False,
on_delete=django.db.models.deletion.CASCADE,
related_name="+",
to="posthog.organization",
),
),
],
),
]
2 changes: 1 addition & 1 deletion products/workflows/backend/migrations/max_migration.txt
Original file line number Diff line number Diff line change
@@ -1 +1 @@
0040_team_workflows_config_frequency_cap_hold
0042_sandbox_sender_organization_block
4 changes: 4 additions & 0 deletions products/workflows/backend/models/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@
from .hog_flow_optimization import HogFlowOptimization
from .hog_flow_revision import HogFlowRevision
from .hog_flow_schedule.hog_flow_schedule import HogFlowSchedule
from .sandbox_sender_organization_block import SandboxSenderOrganizationBlock
from .sandbox_sender_tenant_state import SandboxSenderTenantState
from .team_workflows_config import TeamWorkflowsConfig
from .workflow_idea import WorkflowIdea
from .workflow_import import WorkflowImport
Expand All @@ -16,6 +18,8 @@
"HogFlowRevision",
"HogFlowSchedule",
"HogFlowTemplate",
"SandboxSenderOrganizationBlock",
"SandboxSenderTenantState",
"TeamWorkflowsConfig",
"WorkflowIdea",
"WorkflowImport",
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,15 @@
from django.db import models


class SandboxSenderOrganizationBlock(models.Model):
organization = models.OneToOneField(
"posthog.Organization", on_delete=models.CASCADE, db_constraint=False, related_name="+"
)
reason = models.TextField()
created_by = models.ForeignKey(
"posthog.User", on_delete=models.SET_NULL, null=True, blank=True, db_constraint=False, related_name="+"
)
created_at = models.DateTimeField(auto_now_add=True)

def __str__(self) -> str:
return self.organization.name
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
from django.db import models


class SandboxSenderTenantState(models.Model):
"""Instance-level mirror of the shared sandbox SES tenant; platform state, not tenant data, so no team_id."""

tenant_name = models.CharField(max_length=64, unique=True)
sending_status = models.CharField(max_length=32, blank=True, default="", db_default="")
synced_at = models.DateTimeField(null=True, blank=True)
79 changes: 47 additions & 32 deletions products/workflows/backend/providers/ses.py
Original file line number Diff line number Diff line change
Expand Up @@ -395,6 +395,13 @@
return isinstance(error, ClientError) and error.response["Error"]["Code"] in ("Throttling", "ThrottlingException")


@frozen
class TenantSendingState:
sending_status: str
reputation_impact: str | None
tenant_arn: str | None


class SESProvider:
ses_client: "SESClient"
ses_v2_client: "SESV2Client"
Expand Down Expand Up @@ -469,7 +476,17 @@
provider feedback loops, third-party listings), so this is the authoritative health source;
our own app metrics only provide the per-workflow diagnosis.
"""
tenant_name = self._tenant_name_for_team(team_id)
state = self.get_tenant_sending_state(self._tenant_name_for_team(team_id))
if state is None:
return None
return {
"sending_status": state.sending_status,
"reputation_impact": state.reputation_impact,
"findings": self._open_findings(state.tenant_arn) if state.tenant_arn else [],
}

def get_tenant_sending_state(self, tenant_name: str) -> TenantSendingState | None:
"""Sending status and reputation impact of the named SES tenant, or None when it doesn't exist."""
try:
tenant = self.ses_v2_client.get_tenant(TenantName=tenant_name)["Tenant"]
except ClientError as e:
Expand All @@ -479,40 +496,38 @@

sending_status: str = tenant.get("SendingStatus", "ENABLED")
tenant_arn = tenant.get("TenantArn")
reputation_impact: str | None = None
findings: list[dict[str, Any]] = []
if not tenant_arn:
return TenantSendingState(sending_status=sending_status, reputation_impact=None, tenant_arn=None)

if tenant_arn:
try:
entity = self.ses_v2_client.get_reputation_entity(
ReputationEntityReference=tenant_arn, ReputationEntityType="RESOURCE"
)["ReputationEntity"]
except ClientError as e:
# A tenant with no attributed sends yet has no reputation entity.
if e.response["Error"]["Code"] != "NotFoundException":
raise
entity = {}
reputation_impact = entity.get("ReputationImpact")
entity = self._reputation_entity(tenant_arn)
return TenantSendingState(
# The aggregate folds in both AWS-managed and customer-managed pauses.
sending_status = entity.get("SendingStatusAggregate", sending_status)

# RESOURCE_ARN is the only filter key AWS documents as usable on its own with this
# scoping (see _iter_open_recommendations for why STATUS is filtered locally).
findings.extend(
{
"finding_type": recommendation.get("Type", ""),
"impact": recommendation.get("Impact", "LOW"),
"description": recommendation.get("Description", ""),
"last_updated_at": recommendation.get("LastUpdatedTimestamp"),
}
for recommendation in self._iter_open_recommendations({"RESOURCE_ARN": tenant_arn})
)
sending_status=entity.get("SendingStatusAggregate", sending_status),
reputation_impact=entity.get("ReputationImpact"),
tenant_arn=tenant_arn,
)

return {
"sending_status": sending_status,
"reputation_impact": reputation_impact,
"findings": findings,
}
def _reputation_entity(self, tenant_arn: str) -> Mapping[str, Any]:
try:
return self.ses_v2_client.get_reputation_entity(
ReputationEntityReference=tenant_arn, ReputationEntityType="RESOURCE"
)["ReputationEntity"]
except ClientError as e:
# A tenant with no attributed sends yet has no reputation entity.
if e.response["Error"]["Code"] != "NotFoundException":
raise
return {}

def _open_findings(self, tenant_arn: str) -> list[dict[str, Any]]:
return [
{
"finding_type": recommendation.get("Type", ""),
"impact": recommendation.get("Impact", "LOW"),
"description": recommendation.get("Description", ""),
"last_updated_at": recommendation.get("LastUpdatedTimestamp"),
}
for recommendation in self._iter_open_recommendations({"RESOURCE_ARN": tenant_arn})
]

def _iter_open_recommendations(
self, finding_filter: dict[Any, str] | None, client: "SESV2Client | None" = None
Expand Down Expand Up @@ -675,7 +690,7 @@
"""
self.ses_v2_client.put_email_identity_feedback_attributes(EmailIdentity=domain, EmailForwardingEnabled=False)

def verify_email_domain(self, domain: str, mail_from_subdomain: str, team_id: int) -> EmailDomainVerification:

Check warning on line 693 in products/workflows/backend/providers/ses.py

View workflow job for this annotation

GitHub Actions / Python code quality (depot-ubuntu-24.04)

lint:complexity

`verify_email_domain` has cyclomatic complexity 34 (warn >10)

Check warning on line 693 in products/workflows/backend/providers/ses.py

View workflow job for this annotation

GitHub Actions / Python code quality (depot-ubuntu-24.04)

`verify_email_domain` has cyclomatic complexity 34 (warn >10)
# Validate the domain contains valid characters for a domain name
DOMAIN_REGEX = r"(?i)^([a-z0-9]+(-[a-z0-9]+)*\.)+[a-z]{2,}$"
if not re.match(DOMAIN_REGEX, domain):
Expand Down
5 changes: 3 additions & 2 deletions products/workflows/backend/services/email_health.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
from products.workflows.backend.models.hog_flow.hog_flow import HogFlow
from products.workflows.backend.providers.ses import SESProvider
from products.workflows.backend.services.email_sending_attribution import fold_email_totals_by_flow
from products.workflows.backend.services.email_sending_tier import own_sends, sent_metric_names
from products.workflows.backend.utils.email_sending_tiers import max_email_sending_tier, resolve_team_email_sending_tier

logger = structlog.get_logger(__name__)
Expand Down Expand Up @@ -90,9 +91,9 @@ def team_email_sending_allowance(team_id: int) -> EmailSendingAllowance:

def _team_email_sends_since(team_id: int, after: datetime) -> int:
totals = fetch_app_metric_totals_by_team_and_source(
app_source="hog_flow", name=["email_sent"], after=after, team_ids=[team_id]
app_source="hog_flow", name=sent_metric_names(), after=after, team_ids=[team_id]
)
return sum(counts.get("email_sent", 0) for counts in totals.get(team_id, {}).values())
return sum(own_sends(counts) for counts in totals.get(team_id, {}).values())


AWS_TENANT_REPUTATION_CACHE_SECONDS = 5 * 60
Expand Down
Loading
Loading