feat(exceptions): opt-in Ed25519 signing of $exception events

.sampo/changesets/exception-signing.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: minor
+---
+
+feat(exceptions): add opt-in Ed25519 signing of `$exception` events. Set `enable_exception_signing=True` and provide an Ed25519 private key in `exception_signing_private_key`, then register the matching public key in your PostHog project. The SDK signs each captured exception over a canonical projection of its `$exception_list`, so error-tracking ingestion can verify it genuinely came from your backend (rather than being forged through the public ingest key) and mark it verified. Backend use only — never ship a private key in a browser/mobile app. Requires the new `[exception-signing]` extra (`pip install posthoganalytics[exception-signing]`).

posthog/__init__.py

@@ -338,6 +338,13 @@ def get_tags() -> Dict[str, Any]:
 code_variables_ignore_patterns = DEFAULT_CODE_VARIABLES_IGNORE_PATTERNS
 in_app_modules = None  # type: Optional[list[str]]
 
+# Opt-in Ed25519 signing of $exception events. Set enable_exception_signing=True and provide an
+# Ed25519 private key (PEM) in exception_signing_private_key; register the matching public key in
+# your PostHog project so ingestion can stamp a trusted $exception_verified flag. Backend use only
+# (never ship a private key in a browser/mobile app). Requires the [exception-signing] extra.
+enable_exception_signing = False  # type: bool
+exception_signing_private_key = None  # type: Optional[str]
+
 
 # NOTE - this and following functions take unpacked kwargs because we needed to make
 # it impossible to write `posthog.capture(distinct-id, event-name)` - basically, to enforce
@@ -1103,6 +1110,8 @@ def setup() -> Client:
             code_variables_mask_patterns=code_variables_mask_patterns,
             code_variables_ignore_patterns=code_variables_ignore_patterns,
             in_app_modules=in_app_modules,
+            enable_exception_signing=enable_exception_signing,
+            exception_signing_private_key=exception_signing_private_key,
         )
 
     # Always set in case user changes it. Preserve Client's auto-disabled state

posthog/client.py

@@ -213,6 +213,8 @@ def __init__(
         code_variables_mask_patterns=None,
         code_variables_ignore_patterns=None,
         in_app_modules: list[str] | None = None,
+        enable_exception_signing=False,
+        exception_signing_private_key=None,
         _dedicated_ai_endpoint=False,
     ):
         """
@@ -376,6 +378,25 @@ def __init__(
 
         self._set_before_send(before_send)
 
+        # Opt-in Ed25519 signing of $exception events (parse the key once here).
+        self.enable_exception_signing = enable_exception_signing
+        self._exception_signer = None
+        if enable_exception_signing:
+            if not exception_signing_private_key:
+                # Opted into a security feature but gave us nothing to sign with — warn loudly
+                # rather than silently sending every $exception unsigned.
+                self.log.warning(
+                    "enable_exception_signing is True but no exception_signing_private_key was "
+                    "provided; $exception events will be sent UNSIGNED."
+                )
+            else:
+                from posthog.exception_signing import make_signer
+
+                try:
+                    self._exception_signer = make_signer(exception_signing_private_key)
+                except Exception as e:
+                    self.log.error("Failed to initialise exception signing: %s", e)
+
         if self.enable_exception_autocapture:
             self.exception_capture = ExceptionCapture(self)
 
@@ -1362,6 +1383,15 @@ def _enqueue(self, msg, disable_geoip):
                 self.log.exception(f"Error in before_send callback: {e}")
                 # Continue with the original message if callback fails
 
+        # Sign $exception events last, so the signature covers the final content actually sent
+        # (after any before_send mutation) and a before_send callback can't strip it.
+        if self._exception_signer is not None and msg.get("event") == "$exception":
+            try:
+                self._exception_signer.sign_event(msg)
+            except Exception as e:
+                self.log.exception(f"Error signing exception event: {e}")
+                # Leave the event unsigned rather than dropping it.
+
         self.log.debug("queueing: %s", msg)
 
         # if send is False, return msg as if it was successfully queued

posthog/exception_signing.py

@@ -0,0 +1,139 @@
+"""Opt-in Ed25519 signing of ``$exception`` events.
+
+When a backend service configures an Ed25519 private key, the SDK signs every captured
+``$exception`` event over a canonical projection of its ``$exception_list`` and attaches the
+signature as event properties. PostHog's error-tracking ingestion (cymbal) re-derives the same
+projection, verifies it against the project's registered *public* key, and stamps a trusted
+``$exception_verified`` flag — proving the exception genuinely came from your backend rather
+than being forged through the public ingest key.
+
+The canonical projection is a deliberately small, byte-stable subset of each exception
+(type, message, and each frame's function/filename/lineno/module). It excludes everything
+cymbal mutates during ingestion (in-app flags, absolute paths, source context, the injected
+exception id) and anything float-valued, so the bytes the SDK signs match the bytes cymbal
+verifies. The encoding is explicit length-prefixed binary rather than JSON, to avoid
+cross-language canonicalisation pitfalls (key order, non-ASCII escaping, number formatting).
+
+Requires the optional ``cryptography`` dependency: ``pip install posthoganalytics[exception-signing]``.
+"""
+
+import base64
+import hashlib
+import struct
+from typing import Any, Optional
+
+CANONICAL_MAGIC = b"PHEXC1\n"
+
+SIGNATURE_PROPERTY = "$exception_signature"
+KEY_ID_PROPERTY = "$exception_signature_key_id"
+VERSION_PROPERTY = "$exception_signature_version"
+SIGNATURE_VERSION = 1
+
+
+def _lp(value: Any) -> bytes:
+    """Length-prefixed UTF-8 encoding: u32 big-endian length + bytes. None/missing -> empty."""
+    if value is None:
+        data = b""
+    else:
+        data = str(value).encode("utf-8")
+    return struct.pack(">I", len(data)) + data
+
+
+def build_canonical(exception_list: Any) -> bytes:
+    """Deterministic, length-prefixed encoding of the signable projection of ``$exception_list``.
+
+    Both the SDK (here) and cymbal must produce identical bytes for the same input, so this
+    reads only stable string/int fields and never floats.
+    """
+    out = bytearray(CANONICAL_MAGIC)
+    exceptions = exception_list if isinstance(exception_list, list) else []
+    out += struct.pack(">I", len(exceptions))
+    for exc in exceptions:
+        exc = exc if isinstance(exc, dict) else {}
+        out += _lp(exc.get("type"))
+        out += _lp(exc.get("value"))
+        stacktrace = exc.get("stacktrace")
+        frames = stacktrace.get("frames") if isinstance(stacktrace, dict) else None
+        frames = frames if isinstance(frames, list) else []
+        out += struct.pack(">I", len(frames))
+        for frame in frames:
+            frame = frame if isinstance(frame, dict) else {}
+            out += _lp(frame.get("function"))
+            out += _lp(frame.get("filename"))
+            lineno = frame.get("lineno")
+            out += _lp(lineno if lineno is None else str(lineno))
+            out += _lp(frame.get("module"))
+    return bytes(out)
+
+
+def derive_key_id(public_key_raw: bytes) -> str:
+    """Stable short fingerprint of a raw 32-byte Ed25519 public key.
+
+    Computed identically by the SDK (from the configured private key) and by PostHog (from the
+    registered public key), so a signature's key id resolves to the right stored key.
+    """
+    digest = hashlib.sha256(public_key_raw).digest()
+    return base64.urlsafe_b64encode(digest).decode("ascii").rstrip("=")[:16]
+
+
+class ExceptionSigner:
+    """Holds a parsed Ed25519 private key and signs ``$exception`` events.
+
+    Constructed once at client init from a PEM private key. Raises a clear error if the optional
+    ``cryptography`` dependency is missing or the key isn't Ed25519.
+    """
+
+    def __init__(self, private_key_pem: str):
+        try:
+            from cryptography.hazmat.primitives import serialization
+            from cryptography.hazmat.primitives.asymmetric.ed25519 import (
+                Ed25519PrivateKey,
+            )
+        except ImportError as e:  # pragma: no cover - exercised via install extras
+            raise ImportError(
+                "Exception signing requires the optional 'cryptography' dependency. "
+                "Install it with: pip install posthoganalytics[exception-signing]"
+            ) from e
+
+        key = serialization.load_pem_private_key(
+            private_key_pem.encode("utf-8"), password=None
+        )
+        if not isinstance(key, Ed25519PrivateKey):
+            raise ValueError(
+                "exception_signing_private_key must be an Ed25519 private key (PEM)"
+            )
+
+        from cryptography.hazmat.primitives.serialization import (
+            Encoding,
+            PublicFormat,
+        )
+
+        self._key = key
+        public_raw = key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
+        self.key_id = derive_key_id(public_raw)
+
+    def sign(self, canonical: bytes) -> str:
+        return base64.b64encode(self._key.sign(canonical)).decode("ascii")
+
+    def sign_event(self, event: dict) -> dict:
+        """Attach signature properties to a ``$exception`` event in place; returns it.
+
+        Non-exception events pass through untouched.
+        """
+        if event.get("event") != "$exception":
+            return event
+        properties = event.get("properties")
+        if not isinstance(properties, dict):
+            return event
+        canonical = build_canonical(properties.get("$exception_list"))
+        properties[SIGNATURE_PROPERTY] = self.sign(canonical)
+        properties[KEY_ID_PROPERTY] = self.key_id
+        properties[VERSION_PROPERTY] = SIGNATURE_VERSION
+        return event
+
+
+def make_signer(private_key_pem: Optional[str]) -> Optional[ExceptionSigner]:
+    """Build a signer from a PEM key, or None when no key is configured."""
+    if not private_key_pem:
+        return None
+    return ExceptionSigner(private_key_pem)