diff --git a/posthog/management/commands/start_temporal_worker.py b/posthog/management/commands/start_temporal_worker.py index 125849ad4ab8..df513f080b96 100644 --- a/posthog/management/commands/start_temporal_worker.py +++ b/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 diff --git a/posthog/settings/temporal.py b/posthog/settings/temporal.py index c15cb68d1bb0..6137d60ac11a 100644 --- a/posthog/settings/temporal.py +++ b/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 diff --git a/posthog/temporal/common/exception_signing.py b/posthog/temporal/common/exception_signing.py new file mode 100644 index 000000000000..40a2aba6f18a --- /dev/null +++ b/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 diff --git a/posthog/temporal/common/test_exception_signing.py b/posthog/temporal/common/test_exception_signing.py new file mode 100644 index 000000000000..6c9a610653c1 --- /dev/null +++ b/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