[APPSEC-60450] Add parsed body address to downstream request analysis

lib/datadog/appsec/configuration/settings.rb

@@ -393,6 +393,20 @@ def self.add_settings!(base)
                     end
                   end
                 end
+
+                settings :downstream_body_analysis do
+                  option :sample_rate do |o|
+                    o.type :float
+                    o.env 'DD_API_SECURITY_DOWNSTREAM_BODY_ANALYSIS_SAMPLE_RATE'
+                    o.default 0.5
+                  end
+
+                  option :max_requests do |o|
+                    o.type :int
+                    o.env 'DD_API_SECURITY_MAX_DOWNSTREAM_REQUEST_BODY_ANALYSIS'
+                    o.default 1
+                  end
+                end
               end
 
               option :sca_enabled do |o|

lib/datadog/appsec/context.rb

@@ -1,5 +1,6 @@
 # frozen_string_literal: true
 
+require_relative 'counter_sampler'
 require_relative 'metrics'
 
 module Datadog
@@ -27,6 +28,9 @@ class Context
       #       it's a `Hash`-like structure.
       attr_reader :state
 
+      # Sampler for downstream HTTP request/response body analysis.
+      attr_reader :downstream_body_sampler
+
       class << self
         def activate(context)
           raise ArgumentError, 'not a Datadog::AppSec::Context' unless context.instance_of?(Context)
@@ -51,9 +55,13 @@ def initialize(trace, span, waf_runner)
         @span = span
         @waf_runner = waf_runner
         @metrics = Metrics::Collector.new
+        @downstream_body_sampler = CounterSampler.new(
+          Datadog.configuration.appsec.api_security.downstream_body_analysis.sample_rate
+        )
         @state = {
           events: [],
-          interrupted: false
+          interrupted: false,
+          downstream_body_analyzed_count: 0
         }
       end
 

lib/datadog/appsec/contrib/excon/ssrf_detection_middleware.rb

@@ -5,24 +5,36 @@
 require_relative '../../event'
 require_relative '../../trace_keeper'
 require_relative '../../security_event'
+require_relative '../../utils/http/url_encoded'
+require_relative '../../utils/http/body'
 
 module Datadog
   module AppSec
     module Contrib
       module Excon
         # AppSec Middleware for Excon
         class SSRFDetectionMiddleware < ::Excon::Middleware::Base
+          SAMPLE_BODY_KEY = :__datadog_appsec_sample_downstream_body
+
           def request_call(data)
             context = AppSec.active_context
             return super unless context && AppSec.rasp_enabled?
 
-            timeout = Datadog.configuration.appsec.waf_timeout
+            mark_body_sampling!(data, context: context)
+
+            headers = normalize_headers(data[:headers])
+            # @type var ephemeral_data: ::Datadog::AppSec::Context::input_data
             ephemeral_data = {
               'server.io.net.url' => request_url(data),
               'server.io.net.request.method' => data[:method].to_s.upcase,
-              'server.io.net.request.headers' => normalize_headers(data[:headers])
+              'server.io.net.request.headers' => headers
             }
 
+            if data[SAMPLE_BODY_KEY] && (body = parse_body(data[:body], content_type: headers['content-type']))
+              ephemeral_data['server.io.net.request.body'] = body
+            end
+
+            timeout = Datadog.configuration.appsec.waf_timeout
             result = context.run_rasp(Ext::RASP_SSRF, {}, ephemeral_data, timeout, phase: Ext::RASP_REQUEST_PHASE)
             handle(result, context: context) if result.match?
 
@@ -33,12 +45,18 @@ def response_call(data)
             context = AppSec.active_context
             return super unless context && AppSec.rasp_enabled?
 
-            timeout = Datadog.configuration.appsec.waf_timeout
+            headers = normalize_headers(data.dig(:response, :headers))
+            # @type var ephemeral_data: ::Datadog::AppSec::Context::input_data
             ephemeral_data = {
               'server.io.net.response.status' => data.dig(:response, :status).to_s,
-              'server.io.net.response.headers' => normalize_headers(data.dig(:response, :headers))
+              'server.io.net.response.headers' => headers
             }
 
+            if data[SAMPLE_BODY_KEY] && (body = parse_body(data.dig(:response, :body), content_type: headers['content-type']))
+              ephemeral_data['server.io.net.response.body'] = body
+            end
+
+            timeout = Datadog.configuration.appsec.waf_timeout
             result = context.run_rasp(Ext::RASP_SSRF, {}, ephemeral_data, timeout, phase: Ext::RASP_RESPONSE_PHASE)
             handle(result, context: context) if result.match?
 
@@ -47,6 +65,22 @@ def response_call(data)
 
           private
 
+          def mark_body_sampling!(data, context:)
+            max = Datadog.configuration.appsec.api_security.downstream_body_analysis.max_requests
+            return if context.state[:downstream_body_analyzed_count] >= max
+            return unless context.downstream_body_sampler.sample?
+
+            context.state[:downstream_body_analyzed_count] += 1
+            data[SAMPLE_BODY_KEY] = true
+          end
+
+          def parse_body(body, content_type:)
+            media_type = Utils::HTTP::MediaType.parse(content_type)
+            return unless media_type
+
+            Utils::HTTP::Body.parse(body, media_type: media_type)
+          end
+
           def request_url(data)
             klass = (data[:scheme] == 'https') ? URI::HTTPS : URI::HTTP
             klass.build(host: data[:host], path: data[:path], query: data[:query]).to_s

lib/datadog/appsec/contrib/faraday/ssrf_detection_middleware.rb

@@ -3,24 +3,36 @@
 require_relative '../../event'
 require_relative '../../trace_keeper'
 require_relative '../../security_event'
+require_relative '../../utils/http/media_type'
+require_relative '../../utils/http/body'
 
 module Datadog
   module AppSec
     module Contrib
       module Faraday
         # AppSec SSRF detection Middleware for Faraday
         class SSRFDetectionMiddleware < ::Faraday::Middleware
+          SAMPLE_BODY_KEY = :__datadog_appsec_sample_downstream_body
+
           def call(env)
             context = AppSec.active_context
             return @app.call(env) unless context && AppSec.rasp_enabled?
 
-            timeout = Datadog.configuration.appsec.waf_timeout
+            mark_body_sampling!(env, context: context)
+
+            headers = normalize_headers(env.request_headers)
+            # @type var ephemeral_data: ::Datadog::AppSec::Context::input_data
             ephemeral_data = {
               'server.io.net.url' => env.url.to_s,
               'server.io.net.request.method' => env.method.to_s.upcase,
-              'server.io.net.request.headers' => env.request_headers.transform_keys(&:downcase)
+              'server.io.net.request.headers' => headers
             }
 
+            if env[SAMPLE_BODY_KEY] && (body = parse_body(env.body, content_type: headers['content-type']))
+              ephemeral_data['server.io.net.request.body'] = body
+            end
+
+            timeout = Datadog.configuration.appsec.waf_timeout
             result = context.run_rasp(Ext::RASP_SSRF, {}, ephemeral_data, timeout, phase: Ext::RASP_REQUEST_PHASE)
             handle(result, context: context) if result.match?
 
@@ -30,18 +42,44 @@ def call(env)
           private
 
           def on_complete(env, context:)
-            timeout = Datadog.configuration.appsec.waf_timeout
-
-            response_headers = env.response_headers || {}
+            headers = normalize_headers(env.response_headers)
+            # @type var ephemeral_data: ::Datadog::AppSec::Context::input_data
             ephemeral_data = {
               'server.io.net.response.status' => env.status.to_s,
-              'server.io.net.response.headers' => response_headers.transform_keys(&:downcase)
+              'server.io.net.response.headers' => headers
             }
 
+            if env[SAMPLE_BODY_KEY] && (body = parse_body(env.body, content_type: headers['content-type']))
+              ephemeral_data['server.io.net.response.body'] = body
+            end
+
+            timeout = Datadog.configuration.appsec.waf_timeout
             result = context.run_rasp(Ext::RASP_SSRF, {}, ephemeral_data, timeout, phase: Ext::RASP_RESPONSE_PHASE)
             handle(result, context: context) if result.match?
           end
 
+          def mark_body_sampling!(env, context:)
+            max = Datadog.configuration.appsec.api_security.downstream_body_analysis.max_requests
+            return if context.state[:downstream_body_analyzed_count] >= max
+            return unless context.downstream_body_sampler.sample?
+
+            context.state[:downstream_body_analyzed_count] += 1
+            env[SAMPLE_BODY_KEY] = true
+          end
+
+          def parse_body(body, content_type:)
+            media_type = Utils::HTTP::MediaType.parse(content_type)
+            return unless media_type
+
+            Utils::HTTP::Body.parse(body, media_type: media_type)
+          end
+
+          def normalize_headers(headers)
+            return {} if headers.nil? || headers.empty?
+
+            headers.transform_keys(&:downcase)
+          end
+
           def handle(result, context:)
             AppSec::Event.tag(context, result)
             TraceKeeper.keep!(context.trace) if result.keep?

lib/datadog/appsec/contrib/rack/gateway/request.rb

@@ -23,16 +23,12 @@ def request
             end
 
             def query
-              # Downstream libddwaf expects keys and values to be extractable
-              # separately so we can't use [[k, v], ...]. We also want to allow
-              # duplicate keys, so we use {k => [v, ...], ...} instead, taking into
-              # account that {k => [v1, v2, ...], ...} is possible for duplicate keys.
-              request.query_string.split('&').each.with_object({}) do |e, hash|
-                k, v = e.split('=').map { |s| CGI.unescape(s) }
-                hash[k] ||= []
-
-                hash[k] << v
-              end
+              ::Rack::Utils.parse_query(request.query_string)
+            rescue => e
+              Datadog.logger.debug { "AppSec: Failed to parse request query string: #{e.class}: #{e.message}" }
+              AppSec.telemetry.report(e, description: 'AppSec: Failed to parse request query string')
+
+              {}
             end
 
             def method

lib/datadog/appsec/contrib/rest_client/request_ssrf_detection_patch.rb

@@ -3,6 +3,8 @@
 require_relative '../../event'
 require_relative '../../trace_keeper'
 require_relative '../../security_event'
+require_relative '../../utils/http/media_type'
+require_relative '../../utils/http/body'
 
 module Datadog
   module AppSec
@@ -14,23 +16,36 @@ def execute(&block)
             context = AppSec.active_context
             return super unless context && AppSec.rasp_enabled?
 
-            timeout = Datadog.configuration.appsec.waf_timeout
+            headers = normalize_request_headers
+            # @type var ephemeral_data: ::Datadog::AppSec::Context::input_data
             ephemeral_data = {
               'server.io.net.url' => url,
               'server.io.net.request.method' => method.to_s.upcase,
-              'server.io.net.request.headers' => normalize_request_headers
+              'server.io.net.request.headers' => headers
             }
 
+            sample_body = mark_body_sampling!(context)
+            if sample_body && (body = parse_body(payload.to_s, content_type: headers['content-type']))
+              ephemeral_data['server.io.net.request.body'] = body
+            end
+
+            timeout = Datadog.configuration.appsec.waf_timeout
             result = context.run_rasp(Ext::RASP_SSRF, {}, ephemeral_data, timeout, phase: Ext::RASP_REQUEST_PHASE)
             handle(result, context: context) if result.match?
 
             response = super
 
+            headers = normalize_response_headers(response)
+            # @type var ephemeral_data: ::Datadog::AppSec::Context::input_data
             ephemeral_data = {
               'server.io.net.response.status' => response.code.to_s,
-              'server.io.net.response.headers' => normalize_response_headers(response)
+              'server.io.net.response.headers' => headers
             }
 
+            if sample_body && (body = parse_body(response.body, content_type: headers['content-type']))
+              ephemeral_data['server.io.net.response.body'] = body
+            end
+
             result = context.run_rasp(Ext::RASP_SSRF, {}, ephemeral_data, timeout, phase: Ext::RASP_RESPONSE_PHASE)
             handle(result, context: context) if result.match?
 
@@ -39,6 +54,24 @@ def execute(&block)
 
           private
 
+          def mark_body_sampling!(context)
+            max = Datadog.configuration.appsec.api_security.downstream_body_analysis.max_requests
+            return false if context.state[:downstream_body_analyzed_count] >= max
+            return false unless context.downstream_body_sampler.sample?
+
+            context.state[:downstream_body_analyzed_count] += 1
+            true
+          end
+
+          def parse_body(body, content_type:)
+            return if body.empty?
+
+            media_type = Utils::HTTP::MediaType.parse(content_type)
+            return unless media_type
+
+            Utils::HTTP::Body.parse(body, media_type: media_type)
+          end
+
           # NOTE: Starting version 2.1.0 headers are already normalized via internal
           #       variable `@processed_headers_lowercase`. In case it's available,
           #       we use it to avoid unnecessary transformation.

lib/datadog/appsec/counter_sampler.rb

@@ -0,0 +1,25 @@
+# frozen_string_literal: true
+
+require_relative '../core/knuth_sampler'
+
+module Datadog
+  module AppSec
+    # Sampler that uses an internal counter to make deterministic sampling decisions.
+    #
+    # Each call to {#sample?} increments the counter and uses it as input to
+    # the underlying Knuth multiplicative hash algorithm.
+    #
+    # @api private
+    class CounterSampler
+      def initialize(rate = 1.0)
+        @sampler = Core::KnuthSampler.new(rate)
+        @counter = 0
+      end
+
+      def sample?
+        @counter += 1
+        @sampler.sample?(@counter)
+      end
+    end
+  end
+end

lib/datadog/appsec/utils/http/body.rb

@@ -0,0 +1,38 @@
+# frozen_string_literal: true
+
+require 'json'
+require 'cgi'
+
+require_relative 'url_encoded'
+
+module Datadog
+  module AppSec
+    module Utils
+      module HTTP
+        # Module for handling HTTP body parsing
+        module Body
+          def self.parse(body, media_type:)
+            return if body.nil?
+
+            body.rewind if body.respond_to?(:rewind) # steep:ignore NoMethod
+            # @type var content: ::String?
+            content = body.respond_to?(:read) ? body.read : body # steep:ignore NoMethod, IncompatibleAssignment
+            body.rewind if body.respond_to?(:rewind) # steep:ignore NoMethod
+
+            return if content.nil? || content.empty?
+
+            if media_type.subtype == 'json' || media_type.subtype.end_with?('+json')
+              JSON.parse(content)
+            elsif media_type.subtype == 'x-www-form-urlencoded'
+              URLEncoded.parse(content)
+            end
+          rescue => e
+            AppSec.telemetry.report(e, description: 'AppSec: Failed to parse body')
+
+            nil
+          end
+        end
+      end
+    end
+  end
+end

lib/datadog/appsec/utils/http/media_range.rb

@@ -168,7 +168,8 @@ def <=>(other)
           #
           # returns true if the MediaType is accepted by this MediaRange
           def ===(other)
-            return self === MediaType.new(other) if other.is_a?(::String)
+            return false if other.nil?
+            return self === MediaType.parse(other) if other.is_a?(::String)
 
             type == other.type && subtype == other.subtype && other.parameters.all? { |k, v| parameters[k] == v } ||
               type == other.type && wildcard?(:subtype) ||