feat(temporal): sign captured $exception events with an HMAC attestation

posthog/management/commands/start_temporal_worker.py

@@ -39,6 +39,7 @@
     ACTIVITIES as CLEANUP_PROPDEFS_ACTIVITIES,
     WORKFLOWS as CLEANUP_PROPDEFS_WORKFLOWS,
 )
+from posthog.temporal.common.exception_signing import make_exception_signer
 from posthog.temporal.common.health_server import HealthCheckServer
 from posthog.temporal.common.liveness_tracker import get_liveness_tracker
 from posthog.temporal.common.logger import configure_logger, get_logger
@@ -560,6 +561,15 @@ def handle(self, *args, **options):
 
         tag_queries(kind="temporal")
 
+        # Sign every $exception captured in this worker process so downstream consumers can
+        # verify it genuinely came from our backend (the ingest key is public). Registered
+        # here — once per worker process, before any capture — so it applies to all task
+        # queues; posthoganalytics re-reads before_send on each capture. No-op when unset.
+        if settings.TEMPORAL_EXCEPTION_SIGNING_SECRET:
+            import posthoganalytics
+
+            posthoganalytics.before_send = make_exception_signer(settings.TEMPORAL_EXCEPTION_SIGNING_SECRET)
+
         # Max AI traces span the Django request and the Temporal activity that runs the agent loop.
         # Without the OTel plugin on the worker, every span emitted from an activity is a root span
         # and the conversation trace splits across disconnected pieces. Force-enable for that queue

posthog/settings/temporal.py

@@ -17,6 +17,12 @@
 TEMPORAL_SECRET_KEY: str = os.getenv("TEMPORAL_SECRET_KEY", SECRET_KEY)
 TEMPORAL_FALLBACK_SECRET_KEYS: list[str] = get_list(os.getenv("TEMPORAL_FALLBACK_SECRET_KEYS", "")) or [SECRET_KEY]
 
+# Shared HMAC secret used to sign $exception events captured by temporal workers, so a
+# downstream consumer can verify the exception genuinely originated from our backend (and
+# wasn't forged against the public ingest key). When unset, signing is disabled. Applies to
+# every temporal worker, not a single product.
+TEMPORAL_EXCEPTION_SIGNING_SECRET: str = os.getenv("TEMPORAL_EXCEPTION_SIGNING_SECRET", "")
+
 
 GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS: int | None = get_from_env(
     "GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS", None, optional=True, type_cast=int

posthog/temporal/common/exception_signing.py

@@ -0,0 +1,136 @@
+"""Sign `$exception` events captured by temporal workers.
+
+The PostHog ingest key is public, so a forged `$exception` event reaches Error Tracking and
+any downstream webhook indistinguishably from a real one. Workers therefore attach an HMAC
+signature over a self-contained *attestation* — a small JSON blob carrying the exception's
+type/message/top-frame plus whatever job context is on the event. A consumer that shares the
+secret can verify the attestation and trust only its contents (server-side ingestion mutates
+the reserved ``$exception_*`` fields, but custom properties pass through byte-for-byte).
+
+Wired as a posthoganalytics ``before_send`` hook in ``start_temporal_worker`` so it covers
+every capture path (the temporal interceptor and inline ``capture_exception`` calls) across
+all task queues. Pure stdlib so it is safe inside the workflow sandbox.
+"""
+
+import hmac
+import json
+import hashlib
+import datetime as dt
+from typing import Any, Callable, Optional
+
+import structlog
+
+logger = structlog.get_logger(__name__)
+
+ATTESTATION_PROPERTY = "$temporal_exception_attestation"
+SIGNATURE_PROPERTY = "$temporal_exception_signature"
+
+# Bound the embedded message so a pathological exception can't bloat the event. The consumer
+# only needs enough to triage; this is plenty.
+MAX_MESSAGE_LENGTH = 4096
+
+# Optional job-context keys copied verbatim from the event properties when present. These are
+# set on data-import activities (via ``properties_to_log``) and the temporal interceptor;
+# absent on other workers, in which case they serialize as null.
+_JOB_CONTEXT = {
+    "team_id": "team_id",
+    "run_id": "run_id",
+    "source_id": "source_id",
+    "schema_id": "schema_id",
+    "workflow_run_id": "temporal.workflow.run_id",
+}
+
+
+def _first_exception(properties: dict[str, Any]) -> dict[str, Any]:
+    exception_list = properties.get("$exception_list")
+    if isinstance(exception_list, list) and exception_list and isinstance(exception_list[0], dict):
+        return exception_list[0]
+    return {}
+
+
+def _top_frame(properties: dict[str, Any]) -> Optional[str]:
+    """Best path-like identifier of the first in-app frame, falling back to the first frame.
+
+    At capture time frames carry ``filename``/``abs_path``/``module`` (the resolved ``source``
+    field is added later, server-side), so we record our own value here for the consumer to
+    attribute against.
+    """
+    fallback: Optional[str] = None
+    exception_list = properties.get("$exception_list")
+    if not isinstance(exception_list, list):
+        return None
+    for exc in exception_list:
+        if not isinstance(exc, dict):
+            continue
+        stacktrace = exc.get("stacktrace")
+        frames = stacktrace.get("frames") if isinstance(stacktrace, dict) else None
+        if not isinstance(frames, list):
+            continue
+        for frame in frames:
+            if not isinstance(frame, dict):
+                continue
+            path = frame.get("filename") or frame.get("abs_path") or frame.get("module")
+            if not path:
+                continue
+            if frame.get("in_app") is True:
+                return path
+            if fallback is None:
+                fallback = path
+    return fallback
+
+
+def build_attestation(properties: dict[str, Any], *, captured_at: str) -> dict[str, Any]:
+    """Build the signable attestation from a built ``$exception`` event's properties."""
+    exc = _first_exception(properties)
+    message = exc.get("value")
+    if isinstance(message, str) and len(message) > MAX_MESSAGE_LENGTH:
+        message = message[:MAX_MESSAGE_LENGTH]
+
+    attestation: dict[str, Any] = {
+        "v": 1,
+        "exception_type": exc.get("type"),
+        "message": message if isinstance(message, str) else None,
+        "top_frame": _top_frame(properties),
+        "captured_at": captured_at,
+    }
+    for out_key, prop_key in _JOB_CONTEXT.items():
+        attestation[out_key] = properties.get(prop_key)
+    return attestation
+
+
+def serialize_attestation(attestation: dict[str, Any]) -> str:
+    """Canonical, stable serialization. The consumer signs/verifies this exact string."""
+    return json.dumps(attestation, separators=(",", ":"), sort_keys=True)
+
+
+def sign(secret: str, payload: str) -> str:
+    return hmac.new(secret.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256).hexdigest()
+
+
+def make_exception_signer(secret: str) -> Callable[[dict[str, Any]], Optional[dict[str, Any]]]:
+    """Build a posthoganalytics ``before_send`` hook that signs ``$exception`` events.
+
+    Non-exception events pass through untouched. Any failure is swallowed (and logged) so
+    signing can never drop or break exception capture.
+    """
+
+    def before_send(event: dict[str, Any]) -> Optional[dict[str, Any]]:
+        try:
+            if not isinstance(event, dict) or event.get("event") != "$exception":
+                return event
+            properties = event.get("properties")
+            if not isinstance(properties, dict):
+                return event
+
+            captured_at = event.get("timestamp")
+            if not isinstance(captured_at, str):
+                captured_at = dt.datetime.now(dt.UTC).isoformat()
+
+            attestation = serialize_attestation(build_attestation(properties, captured_at=captured_at))
+            properties[ATTESTATION_PROPERTY] = attestation
+            properties[SIGNATURE_PROPERTY] = sign(secret, attestation)
+        except Exception:
+            logger.exception("Failed to sign temporal exception event")
+        return event
+
+    return before_send

posthog/temporal/common/test_exception_signing.py

@@ -0,0 +1,167 @@
+import hmac
+import json
+import hashlib
+
+import pytest
+
+from posthog.temporal.common.exception_signing import (
+    ATTESTATION_PROPERTY,
+    MAX_MESSAGE_LENGTH,
+    SIGNATURE_PROPERTY,
+    build_attestation,
+    make_exception_signer,
+    serialize_attestation,
+    sign,
+)
+
+SECRET = "test-secret-0123456789abcdef"
+CAPTURED_AT = "2026-06-10T12:00:00+00:00"
+
+
+def _exception_event(properties: dict | None = None) -> dict:
+    props = {
+        "$exception_list": [
+            {
+                "type": "HTTPError",
+                "value": "401 Client Error: Unauthorized for url: https://api.stripe.com/v1/charges",
+                "stacktrace": {
+                    "frames": [
+                        {"module": "requests.models", "filename": "requests/models.py", "in_app": False},
+                        {
+                            "module": "posthog.temporal.data_imports.sources.stripe.source",
+                            "filename": "posthog/temporal/data_imports/sources/stripe/source.py",
+                            "in_app": True,
+                        },
+                    ]
+                },
+            }
+        ],
+        "team_id": 12345,
+        "run_id": "0196f1a2-1111-7000-8000-0123456789ab",
+        "source_id": "0196f1a2-3333-7000-8000-0123456789ab",
+        "schema_id": "0196f1a2-4444-7000-8000-0123456789ab",
+        "temporal.workflow.run_id": "11111111-2222-4333-8444-555555555555",
+    }
+    if properties:
+        props.update(properties)
+    return {"event": "$exception", "timestamp": CAPTURED_AT, "properties": props}
+
+
+class TestBuildAttestation:
+    def test_extracts_content_and_job_context(self):
+        att = build_attestation(_exception_event()["properties"], captured_at=CAPTURED_AT)
+        assert att == {
+            "v": 1,
+            "exception_type": "HTTPError",
+            "message": "401 Client Error: Unauthorized for url: https://api.stripe.com/v1/charges",
+            "top_frame": "posthog/temporal/data_imports/sources/stripe/source.py",
+            "captured_at": CAPTURED_AT,
+            "team_id": 12345,
+            "run_id": "0196f1a2-1111-7000-8000-0123456789ab",
+            "source_id": "0196f1a2-3333-7000-8000-0123456789ab",
+            "schema_id": "0196f1a2-4444-7000-8000-0123456789ab",
+            "workflow_run_id": "11111111-2222-4333-8444-555555555555",
+        }
+
+    def test_prefers_first_in_app_frame_over_first_frame(self):
+        att = build_attestation(_exception_event()["properties"], captured_at=CAPTURED_AT)
+        assert att["top_frame"] == "posthog/temporal/data_imports/sources/stripe/source.py"
+
+    def test_falls_back_to_first_frame_when_no_in_app(self):
+        props = _exception_event()["properties"]
+        for frame in props["$exception_list"][0]["stacktrace"]["frames"]:
+            frame["in_app"] = False
+        att = build_attestation(props, captured_at=CAPTURED_AT)
+        assert att["top_frame"] == "requests/models.py"
+
+    def test_truncates_long_message(self):
+        long = "x" * (MAX_MESSAGE_LENGTH + 500)
+        props = _exception_event({"$exception_list": [{"type": "E", "value": long}]})["properties"]
+        att = build_attestation(props, captured_at=CAPTURED_AT)
+        assert att["message"] == "x" * MAX_MESSAGE_LENGTH
+
+    @pytest.mark.parametrize(
+        "props,expected_type,expected_message,expected_frame",
+        [
+            ({"$exception_list": []}, None, None, None),
+            ({"$exception_list": [{"type": "KeyError"}]}, "KeyError", None, None),
+            ({"$exception_list": [{"value": "boom"}]}, None, "boom", None),
+            ({}, None, None, None),
+        ],
+    )
+    def test_tolerates_missing_fields(self, props, expected_type, expected_message, expected_frame):
+        att = build_attestation(props, captured_at=CAPTURED_AT)
+        assert att["exception_type"] == expected_type
+        assert att["message"] == expected_message
+        assert att["top_frame"] == expected_frame
+        # Absent job context serializes as null, not missing.
+        assert att["team_id"] is None and att["run_id"] is None
+
+
+class TestSign:
+    def test_serialization_is_deterministic_and_sorted(self):
+        att = build_attestation(_exception_event()["properties"], captured_at=CAPTURED_AT)
+        s = serialize_attestation(att)
+        assert s == serialize_attestation(att)
+        # sort_keys → "captured_at" precedes "exception_type"
+        assert s.index('"captured_at"') < s.index('"exception_type"')
+        assert " " not in s  # compact separators
+
+    def test_signature_reproduces_with_same_secret(self):
+        payload = serialize_attestation(build_attestation(_exception_event()["properties"], captured_at=CAPTURED_AT))
+        expected = hmac.new(SECRET.encode(), payload.encode(), hashlib.sha256).hexdigest()
+        assert sign(SECRET, payload) == expected
+
+    def test_different_secret_or_payload_changes_signature(self):
+        payload = serialize_attestation(build_attestation(_exception_event()["properties"], captured_at=CAPTURED_AT))
+        assert sign(SECRET, payload) != sign("other-secret", payload)
+        assert sign(SECRET, payload) != sign(SECRET, payload + "x")
+
+
+class TestBeforeSendHook:
+    def test_signs_exception_event(self):
+        hook = make_exception_signer(SECRET)
+        event = hook(_exception_event())
+
+        attestation = event["properties"][ATTESTATION_PROPERTY]
+        signature = event["properties"][SIGNATURE_PROPERTY]
+        # The signature verifies against the exact embedded string.
+        assert sign(SECRET, attestation) == signature
+        # And the embedded attestation parses to the trusted content.
+        parsed = json.loads(attestation)
+        assert parsed["message"].startswith("401 Client Error")
+        assert parsed["team_id"] == 12345
+
+    def test_passes_through_non_exception_events_untouched(self):
+        hook = make_exception_signer(SECRET)
+        event = {"event": "schema non-retryable error", "properties": {"sourceType": "Stripe"}}
+        assert hook(event) == event
+        assert ATTESTATION_PROPERTY not in event["properties"]
+
+    def test_uses_event_timestamp_as_captured_at(self):
+        hook = make_exception_signer(SECRET)
+        event = hook(_exception_event())
+        assert json.loads(event["properties"][ATTESTATION_PROPERTY])["captured_at"] == CAPTURED_AT
+
+    @pytest.mark.parametrize(
+        "event",
+        [
+            {"event": "$exception"},  # no properties
+            {"event": "$exception", "properties": None},
+            {"event": "$exception", "properties": {}},
+            {"properties": {}},  # no event name
+            {},
+        ],
+    )
+    def test_never_raises_on_malformed_events(self, event):
+        hook = make_exception_signer(SECRET)
+        # Should return the event (possibly signed, possibly untouched) without raising.
+        assert hook(event) is event
+
+    def test_tampering_with_message_breaks_verification(self):
+        hook = make_exception_signer(SECRET)
+        event = hook(_exception_event())
+        signature = event["properties"][SIGNATURE_PROPERTY]
+        # An attacker swaps the displayed message but keeps the (now stale) signature.
+        tampered = event["properties"][ATTESTATION_PROPERTY].replace("401 Client Error", "RUN rm -rf /")
+        assert sign(SECRET, tampered) != signature