feat(webhooks): verify_and_parse_* API for compressed payloads (CHA-3071)

docs/webhooks/webhooks_overview/webhooks_overview.md

@@ -96,6 +96,85 @@ All webhook requests contain these headers:
 | X-Api-Key         | Your application’s API key. Should be used to validate request signature                                             | a1b23cdefgh4                                                     |
 | X-Signature       | HMAC signature of the request body. See Signature section                                                            | ca978112ca1bbdcafac231b39a23dc4da786eff8147c4e72b9807785afee48bb |
 
+## Compressed webhook bodies
+
+GZIP compression can be enabled for hooks payloads from the Dashboard. Enabling compression reduces the payload size significantly (often 70–90% smaller) reducing your bandwidth usage on Stream. The computation overhead introduced by the decompression step is usually negligible and offset by the much smaller payload.
+
+When payload compression is enabled, webhook HTTP requests will include the `Content-Encoding: gzip` header and the request body will be compressed with GZIP. Some HTTP servers and middleware (Rails, Django, Laravel, Spring Boot, ASP.NET) handle this transparently and strip the header before your handler runs — in that case the body you see is already raw JSON.
+
+Before enabling compression, make sure that:
+
+* Your backend integration is using a recent version of our official SDKs with compression support
+* If you don't use an official SDK, make sure that your code supports receiving compressed payloads
+* The payload signature check is done on the **uncompressed** payload
+
+### Decoding a compressed webhook in Ruby (Rails)
+
+The Ruby SDK exposes three composite helpers on `StreamChat::Client`, one per transport, so you do not have to wire `Zlib`, `Base64`, and `OpenSSL::HMAC` together yourself. Each method decodes the body, verifies the `X-Signature` HMAC against the **uncompressed** JSON, parses the JSON, and returns the event as a `Hash`. Any failure (bad signature, malformed gzip, malformed base64) raises `StreamChat::WebhookSignatureError`.
+
+* `verify_and_parse_webhook(body, x_signature)` — HTTP webhooks.
+* `verify_and_parse_sqs(message_body, x_signature)` — SQS firehose messages (base64 + optional gzip).
+* `verify_and_parse_sns(message, x_signature)` — SNS notification messages (base64 + optional gzip).
+
+GZIP is detected from the body bytes (the `1f 8b` magic prefix), not from the `Content-Encoding` header, so the same handler works whether or not your HTTP server transparently decompresses the request body before it reaches you.
+
+```ruby
+require 'stream-chat'
+
+STREAM_CLIENT = StreamChat::Client.new('STREAM_KEY', 'STREAM_SECRET')
+
+class WebhooksController < ApplicationController
+  skip_before_action :verify_authenticity_token
+
+  def stream
+    body = request.raw_post # binary safe; do NOT use params
+
+    event = STREAM_CLIENT.verify_and_parse_webhook(
+      body,
+      request.headers['X-Signature']
+    )
+
+    Rails.logger.info("Stream webhook: #{event['type']}")
+    head :ok
+  rescue StreamChat::WebhookSignatureError => e
+    Rails.logger.warn("Rejected Stream webhook: #{e.message}")
+    head :unauthorized
+  end
+end
+```
+
+If you also want to support the legacy "I just want to check the signature myself" flow, `client.verify_webhook(body, x_signature)` is still available and returns a boolean. It does **not** handle compression — use `verify_and_parse_webhook` for new integrations.
+
+### Decoding a compressed SQS / SNS firehose message
+
+SQS and SNS message bodies must be valid UTF-8, so when GZIP compression is enabled the gzipped bytes are additionally **base64-wrapped** before being placed on the queue. `verify_and_parse_sqs` (and its SNS twin) unwrap the queue envelope before verifying the signature, so the same call works whether or not compression is enabled:
+
+```ruby
+require 'aws-sdk-sqs'
+require 'stream-chat'
+
+client = StreamChat::Client.new('STREAM_KEY', 'STREAM_SECRET')
+sqs = Aws::SQS::Client.new
+
+resp = sqs.receive_message(queue_url: ENV.fetch('STREAM_SQS_URL'), max_number_of_messages: 10)
+
+resp.messages.each do |msg|
+  signature = msg.message_attributes&.dig('X-Signature', :string_value)
+
+  event = client.verify_and_parse_sqs(msg.body, signature)
+  # ...handle event...
+
+  sqs.delete_message(queue_url: ENV.fetch('STREAM_SQS_URL'), receipt_handle: msg.receipt_handle)
+rescue StreamChat::WebhookSignatureError => e
+  # do NOT delete the message - leave it for the DLQ / retry policy
+  Rails.logger.warn("Rejected Stream SQS message: #{e.message}")
+end
+```
+
+The exact attribute name that carries the signature may vary — refer to the SQS / SNS pages in this section for the up-to-date list. The decoding rules themselves do not change: the signature is always computed over the **uncompressed** JSON.
+
+If you need to build a custom pipeline, the module-level primitives are also exposed: `StreamChat::Webhook.ungzip_payload`, `decode_sqs_payload`, `decode_sns_payload`, `verify_signature`, and `parse_event`. The composite `verify_and_parse_*` methods are thin wrappers on top of these.
+
 ## Webhook types
 
 In addition to the above there are 3 special webhooks.

lib/stream-chat/client.rb

@@ -1,7 +1,11 @@
 # typed: strict
 # frozen_string_literal: true
 
+require 'base64'
 require 'open-uri'
+require 'openssl'
+require 'stringio'
+require 'zlib'
 require 'faraday'
 require 'faraday/multipart'
 require 'faraday/net_http_persistent'
@@ -18,6 +22,7 @@
 require 'stream-chat/types'
 require 'stream-chat/moderation'
 require 'stream-chat/channel_batch_updater'
+require 'stream-chat/webhook'
 
 module StreamChat
   DEFAULT_BLOCKLIST = 'profanity'
@@ -698,11 +703,67 @@ def get_rate_limits(server_side: false, android: false, ios: false, web: false,
       get('rate_limits', params: params)
     end
 
-    # Verify the signature added to a webhook event.
+    # Verify the signature on a webhook event using the client's API secret.
+    #
+    # Backward-compatible boolean helper. New integrations should call
+    # {#verify_and_parse_webhook} (or the SQS / SNS variants), which also handle
+    # gzip payload compression and return the parsed event.
     sig { params(request_body: String, x_signature: String).returns(T::Boolean) }
     def verify_webhook(request_body, x_signature)
-      signature = OpenSSL::HMAC.hexdigest('SHA256', @api_secret, request_body)
-      signature == x_signature
+      StreamChat::Webhook.verify_signature(request_body, x_signature, @api_secret)
+    end
+
+    # Verify and parse an HTTP webhook event.
+    #
+    # Decompresses `body` when gzipped (detected from the body bytes), verifies
+    # the `X-Signature` header against the client's API secret, and returns the
+    # parsed event as a `Hash`. Typed event classes are planned for a future
+    # release.
+    #
+    # @param body [String, Array<Integer>] Raw HTTP request body bytes Stream
+    #   signed.
+    # @param signature [String] Value of the `X-Signature` header.
+    # @return [Hash] The parsed event.
+    # @raise [WebhookSignatureError] When the signature does not match or the
+    #   gzip envelope is malformed.
+    sig do
+      params(
+        body: T.any(String, T::Array[Integer]),
+        signature: String
+      ).returns(T::Hash[String, T.untyped])
+    end
+    def verify_and_parse_webhook(body, signature)
+      StreamChat::Webhook.verify_and_parse_webhook(body, signature, @api_secret)
+    end
+
+    # Verify and parse an SQS firehose webhook event.
+    #
+    # Reverses the base64 (+ optional gzip) wrapping on the SQS `Body`,
+    # verifies the `X-Signature` message attribute against the client's API
+    # secret, and returns the parsed event.
+    #
+    # @param message_body [String] SQS message `Body` string.
+    # @param signature [String] Value of the `X-Signature` message attribute.
+    # @return [Hash] The parsed event.
+    # @raise [WebhookSignatureError]
+    sig { params(message_body: String, signature: String).returns(T::Hash[String, T.untyped]) }
+    def verify_and_parse_sqs(message_body, signature)
+      StreamChat::Webhook.verify_and_parse_sqs(message_body, signature, @api_secret)
+    end
+
+    # Verify and parse an SNS firehose webhook event.
+    #
+    # Reverses the base64 (+ optional gzip) wrapping on the SNS notification
+    # `Message`, verifies the `X-Signature` message attribute against the
+    # client's API secret, and returns the parsed event.
+    #
+    # @param message [String] SNS notification `Message` field.
+    # @param signature [String] Value of the `X-Signature` message attribute.
+    # @return [Hash] The parsed event.
+    # @raise [WebhookSignatureError]
+    sig { params(message: String, signature: String).returns(T::Hash[String, T.untyped]) }
+    def verify_and_parse_sns(message, signature)
+      StreamChat::Webhook.verify_and_parse_sns(message, signature, @api_secret)
     end
 
     # Allows you to send custom events to a connected user.

lib/stream-chat/errors.rb

@@ -47,4 +47,21 @@ def to_s
   end
 
   class StreamChannelException < StandardError; end
+
+  # Raised when a webhook payload cannot be decoded, decompressed, or its
+  # signature does not match the expected HMAC. This is a local verification
+  # failure with no HTTP response attached, so it inherits directly from
+  # `StandardError` rather than `StreamAPIException`.
+  class WebhookSignatureError < StandardError
+    extend T::Sig
+
+    sig { returns(String) }
+    attr_reader :reason
+
+    sig { params(reason: String).void }
+    def initialize(reason)
+      @reason = T.let(reason, String)
+      super
+    end
+  end
 end

lib/stream-chat/webhook.rb

@@ -0,0 +1,194 @@
+# typed: strict
+# frozen_string_literal: true
+
+require 'base64'
+require 'json'
+require 'openssl'
+require 'sorbet-runtime'
+require 'stringio'
+require 'zlib'
+
+require 'stream-chat/errors'
+
+module StreamChat
+  # Stateless helpers implementing the cross-SDK webhook contract documented at
+  # https://getstream.io/chat/docs/node/webhooks_overview/.
+  #
+  # The composite functions (`verify_and_parse_webhook`, `verify_and_parse_sqs`,
+  # `verify_and_parse_sns`) are the recommended entry points. The primitives
+  # they compose (`ungzip_payload`, `decode_sqs_payload`, `decode_sns_payload`,
+  # `verify_signature`, `parse_event`) are exposed so callers can build custom
+  # flows or run individual steps in isolation.
+  #
+  # The Ruby SDK currently returns the parsed JSON as a `Hash`; typed event
+  # classes will land in a future release.
+  module Webhook # rubocop:disable Metrics/ModuleLength
+    extend T::Sig
+
+    GZIP_MAGIC = T.let("\x1f\x8b".b.freeze, String)
+
+    # Coerces the webhook body into a binary `String` regardless of whether
+    # the caller hands us a `String` (HTTP `request.raw_post`) or an array of
+    # bytes (which is what some Ruby SQS clients yield when the message body
+    # is binary safe).
+    sig { params(body: T.any(String, T::Array[Integer])).returns(String) }
+    def self.normalize_body(body)
+      raw =
+        if body.is_a?(Array)
+          body.pack('C*')
+        else
+          String.new(body)
+        end
+      raw.force_encoding(Encoding::ASCII_8BIT)
+    end
+
+    # Returns `body` unchanged unless it starts with the gzip magic
+    # (`1f 8b`, per RFC 1952), in which case the gzip stream is inflated and
+    # the decompressed bytes are returned.
+    #
+    # Magic-byte detection (rather than relying on a header) keeps the same
+    # handler correct when middleware - Rack, Rails - auto-decompresses the
+    # request before your code sees it.
+    sig { params(body: T.any(String, T::Array[Integer])).returns(String) }
+    def self.ungzip_payload(body)
+      raw = normalize_body(body)
+      return raw unless raw.start_with?(GZIP_MAGIC)
+
+      begin
+        Zlib::GzipReader.new(StringIO.new(raw)).read.force_encoding(Encoding::ASCII_8BIT)
+      rescue Zlib::Error => e
+        raise WebhookSignatureError, "failed to decompress gzip payload: #{e.message}"
+      end
+    end
+
+    # Reverses the SQS firehose envelope: the message `Body` is base64-decoded
+    # and, when the result begins with the gzip magic, gzip-decompressed. The
+    # same call works whether or not Stream is currently compressing payloads.
+    sig { params(body: String).returns(String) }
+    def self.decode_sqs_payload(body)
+      decoded =
+        begin
+          Base64.strict_decode64(body)
+        rescue ArgumentError => e
+          raise WebhookSignatureError, "failed to base64-decode payload: #{e.message}"
+        end
+      ungzip_payload(decoded)
+    end
+
+    # Identical to `decode_sqs_payload`; exposed under both names so call sites
+    # read intent.
+    sig { params(message: String).returns(String) }
+    def self.decode_sns_payload(message)
+      decode_sqs_payload(message)
+    end
+
+    # Constant-time HMAC-SHA256 verification of `signature` against the digest
+    # of `body` keyed by `secret`.
+    #
+    # The signature is always computed over the **uncompressed** JSON bytes,
+    # so callers that decoded a gzipped or base64-wrapped payload must pass
+    # the inflated bytes here.
+    sig do
+      params(
+        body: T.any(String, T::Array[Integer]),
+        signature: String,
+        secret: String
+      ).returns(T::Boolean)
+    end
+    def self.verify_signature(body, signature, secret)
+      raw = normalize_body(body)
+      expected = OpenSSL::HMAC.hexdigest('SHA256', secret, raw)
+      constant_time_equal?(expected, signature)
+    end
+
+    # Parse a JSON-encoded webhook event into a `Hash`.
+    #
+    # The Ruby SDK currently returns the parsed JSON as a `Hash`; typed event
+    # classes will land in a future release. The function name matches the
+    # documented primitive so callers can swap in a typed parser later without
+    # changing call sites.
+    sig { params(payload: String).returns(T::Hash[String, T.untyped]) }
+    def self.parse_event(payload)
+      result = JSON.parse(payload)
+      raise WebhookSignatureError, 'failed to parse webhook event: top-level value is not an object' unless result.is_a?(Hash)
+
+      result
+    end
+
+    sig do
+      params(
+        payload: String,
+        signature: String,
+        secret: String
+      ).returns(T::Hash[String, T.untyped])
+    end
+    def self.verify_and_parse_internal(payload, signature, secret)
+      raise WebhookSignatureError, 'invalid webhook signature' unless verify_signature(payload, signature, secret)
+
+      parse_event(payload)
+    end
+    private_class_method :verify_and_parse_internal
+
+    # Decompress `body` when gzipped, verify the HMAC `signature`, and return
+    # the parsed event.
+    sig do
+      params(
+        body: T.any(String, T::Array[Integer]),
+        signature: String,
+        secret: String
+      ).returns(T::Hash[String, T.untyped])
+    end
+    def self.verify_and_parse_webhook(body, signature, secret)
+      verify_and_parse_internal(ungzip_payload(body), signature, secret)
+    end
+
+    # Decode the SQS `Body` (base64, then gzip-if-magic), verify the HMAC
+    # `signature` from the `X-Signature` message attribute, and return the
+    # parsed event.
+    sig do
+      params(
+        message_body: String,
+        signature: String,
+        secret: String
+      ).returns(T::Hash[String, T.untyped])
+    end
+    def self.verify_and_parse_sqs(message_body, signature, secret)
+      verify_and_parse_internal(decode_sqs_payload(message_body), signature, secret)
+    end
+
+    # Decode the SNS notification `Message` (identical to SQS handling), verify
+    # the HMAC `signature` from the `X-Signature` message attribute, and return
+    # the parsed event.
+    sig do
+      params(
+        message: String,
+        signature: String,
+        secret: String
+      ).returns(T::Hash[String, T.untyped])
+    end
+    def self.verify_and_parse_sns(message, signature, secret)
+      verify_and_parse_internal(decode_sns_payload(message), signature, secret)
+    end
+
+    # Timing-safe equality check used to compare the locally computed HMAC
+    # with the `X-Signature` header. Prefers the OpenSSL primitive when the
+    # Ruby build exposes it, otherwise falls back to a manual byte XOR loop
+    # that does not short-circuit on the first mismatch.
+    sig { params(left: String, right: String).returns(T::Boolean) }
+    def self.constant_time_equal?(left, right)
+      a = left.b
+      b = right.b
+      return false unless a.bytesize == b.bytesize
+
+      if OpenSSL.respond_to?(:fixed_length_secure_compare)
+        OpenSSL.fixed_length_secure_compare(a, b)
+      else
+        a_bytes = a.bytes
+        b_bytes = b.bytes
+        diff = 0
+        a_bytes.each_with_index { |byte, i| diff |= byte ^ b_bytes[i] }
+        diff.zero?
+      end
+    end
+  end
+end