defmodule Accrue.Webhook.Plug do @moduledoc """ Core webhook ingestion plug. Processes incoming webhook requests by: 1. Extracting the raw body from `conn.assigns[:raw_body]` (populated by `Accrue.Webhook.CachingBodyReader` in the pipeline) 2. Verifying the `Stripe-Signature` header via `Accrue.Webhook.Signature` 3. Projecting the verified event into `%Accrue.Webhook.Event{}` 4. Storing the verified event and raw body in `conn.private` for downstream processing Signature failures raise `Accrue.SignatureError`, rescued to HTTP 400. ## Transactional pipeline After verification, `Accrue.Webhook.Ingest.run/4` atomically persists the webhook event, enqueues an Oban dispatch job, and records an accrue_events ledger entry -- all in a single `Ecto.Multi` transaction. """ @behaviour Plug import Plug.Conn require Logger alias Accrue.Webhook.Signature @impl true def init(opts), do: opts @impl true def call(conn, opts) do processor = Keyword.fetch!(opts, :processor) endpoint = Keyword.get(opts, :endpoint) :telemetry.span( [:accrue, :webhook, :receive], %{processor: processor, endpoint: endpoint}, fn -> result = do_call(conn, processor, endpoint) {result, %{processor: processor, endpoint: endpoint}} end ) rescue e in Accrue.SignatureError -> Logger.warning("Webhook signature verification failed: #{e.reason}") conn |> send_resp(400, Jason.encode!(%{error: "signature_verification_failed"})) |> halt() e in Accrue.ConfigError -> Logger.error("Webhook setup error:\n" <> Exception.message(e)) conn |> send_resp(500, Jason.encode!(%{error: "internal_server_error"})) |> halt() end defp do_call(conn, :stripe, endpoint) do raw_body = flatten_raw_body(conn) sig_header = get_req_header(conn, "stripe-signature") |> List.first() unless sig_header do raise Accrue.SignatureError, reason: "missing stripe-signature header" end secrets = resolve_secrets!(endpoint, :stripe) stripe_event = Signature.verify!(raw_body, sig_header, secrets) # Transactional persist + Oban enqueue. Accrue.Webhook.Ingest.run(conn, :stripe, stripe_event, raw_body, endpoint) end defp do_call(conn, :braintree, endpoint) do conn = Plug.Conn.fetch_query_params(conn) bt_signature = conn.params["bt_signature"] bt_payload = conn.params["bt_payload"] unless bt_signature do raise Accrue.SignatureError, reason: "missing bt_signature" end unless bt_payload do raise Accrue.SignatureError, reason: "missing bt_payload" end braintree_event = Signature.parse_braintree!(bt_signature, bt_payload) # Braintree webhooks lack a unique event ID, so we hash the payload for idempotency. # We construct a %LatticeStripe.Event{} simply to satisfy the Ingest.run signature and DB constraints. event_id = "bt_" <> Base.encode16(:crypto.hash(:sha256, bt_payload), case: :lower) kind = braintree_event["kind"] || "unknown" livemode = Braintree.get_env(:environment) == :production subject = braintree_event["subject"] || %{} subject_entity = subject |> Map.values() |> Enum.find(&is_map/1) || %{} object_id = subject_entity["id"] object_data = if object_id, do: Map.put(subject, "id", object_id), else: subject pseudo_event = %LatticeStripe.Event{ id: event_id, type: kind, livemode: livemode, data: %{"object" => object_data} } Accrue.Webhook.Ingest.run(conn, :braintree, pseudo_event, bt_payload, endpoint) end # Endpoint-aware secret resolution. # # Multi-endpoint mode: when `endpoint:` is passed via the plug init opts, # look up `[:webhook_endpoints, endpoint, :secret]` from `Accrue.Config`. # If the endpoint is missing, raise `Accrue.SignatureError` (rescued to 400) # — fail closed, never bypass verification. # # Legacy mode: when no `endpoint:` is set, fall back to # `Accrue.Config.webhook_signing_secrets/1` keyed by processor atom. defp resolve_secrets!(nil, processor), do: Accrue.Config.webhook_signing_secrets(processor) defp resolve_secrets!(endpoint, _processor) when is_atom(endpoint) do endpoints = Accrue.Config.webhook_endpoints() case Keyword.get(endpoints, endpoint) do nil -> raise Accrue.SignatureError, reason: "no webhook_endpoints config for endpoint #{inspect(endpoint)}" cfg when is_list(cfg) -> case Keyword.fetch(cfg, :secret) do {:ok, secret} when is_binary(secret) and secret != "" -> secret {:ok, secrets} when is_list(secrets) and secrets != [] -> secrets _ -> raise Accrue.SignatureError, reason: "no :secret in webhook_endpoints[#{inspect(endpoint)}]" end end end @doc false def flatten_raw_body(conn) do case conn.assigns[:raw_body] do nil -> diagnostic = Accrue.SetupDiagnostic.webhook_raw_body( details: "Expected conn.assigns[:raw_body]; configure body_reader: {Accrue.Webhook.CachingBodyReader, :read_body, []}" ) raise Accrue.ConfigError, key: :webhook_signing_secrets, diagnostic: diagnostic chunks when is_list(chunks) -> chunks |> Enum.reverse() |> IO.iodata_to_binary() binary when is_binary(binary) -> binary end end end