Add concurrent request rate limiter and optimize middleware stack

config/cloud_controller.yml

@@ -275,6 +275,11 @@ rate_limiter_v2_api:
   global_admin_limit: 20000
   reset_interval_in_minutes: 60
 
+concurrency_rate_limiter:
+  enabled: false
+  blocking_limit: 10
+  logging_limit: 10
+
 temporary_enable_v2: true
 
 max_concurrent_service_broker_requests: 0

errors/v2.yml

@@ -128,6 +128,17 @@
   http_code: 422
   message: "The %s is being deleted"
 
+10021:
+  name: ConcurrentRequestLimitExceeded
+  http_code: 429
+  message: "Too many concurrent requests. Please retry."
+
+10022:
+  name: IPBasedConcurrentRequestLimitExceeded
+  http_code: 429
+  message: "Too many concurrent requests from this IP. Please retry."
+
+
 20001:
   name: UserInvalid
   http_code: 400

lib/cloud_controller/config.rb

@@ -64,9 +64,18 @@ def merge_defaults(orig_config)
         config = orig_config.dup
         config[:db] ||= {}
         ensure_config_has_database_parts(config)
+        apply_redis_counter_defaults(config)
         sanitize(config)
       end
 
+      def apply_redis_counter_defaults(config)
+        return unless config.dig(:concurrency_rate_limiter, :enabled) ||
+                      (config[:max_concurrent_service_broker_requests] || 0) > 0
+
+        config[:redis_connection_pool_size] = config.dig(:puma, :max_threads) || 1
+        config[:redis_counter_ttl_seconds] = config[:request_timeout_in_seconds] + 1
+      end
+
       def ensure_config_has_database_parts(config)
         abort_no_db_connection! if ENV['DB_CONNECTION_STRING'].nil? && config[:db][:database].nil?
         config[:db][:db_connection_string] ||= ENV.fetch('DB_CONNECTION_STRING', nil)

lib/cloud_controller/config_schemas/api_schema.rb

@@ -384,6 +384,8 @@ class ApiSchema < VCAP::Config
             reset_interval_in_minutes: Integer
           },
           max_concurrent_service_broker_requests: Integer,
+          optional(:redis_connection_pool_size) => Integer,
+          optional(:redis_counter_ttl_seconds) => Integer,
           shared_isolation_segment_name: String,
 
           optional(:rate_limiter_v2_api) => {
@@ -395,6 +397,12 @@ class ApiSchema < VCAP::Config
             reset_interval_in_minutes: Integer
           },
 
+          optional(:concurrency_rate_limiter) => {
+            enabled: bool,
+            optional(:blocking_limit) => Integer,
+            optional(:logging_limit) => Integer
+          },
+
           optional(:temporary_enable_v2) => bool,
 
           allow_app_ssh_access: bool,

lib/cloud_controller/rack_app_builder.rb

@@ -7,28 +7,40 @@
 require 'rate_limiter'
 require 'service_broker_rate_limiter'
 require 'rate_limiter_v2_api'
+require 'concurrency_rate_limiter'
 require 'new_relic_custom_attributes'
 require 'zipkin'
 require 'block_v3_only_roles'
 require 'below_min_cli_warning'
+require 'user_context_setter'
 
 module VCAP::CloudController
   class RackAppBuilder
+    # rubocop:disable Metrics/MethodLength, Metrics/BlockLength
     def build(config, request_metrics, request_logs)
       token_decoder = VCAP::CloudController::UaaTokenDecoder.new(config.get(:uaa))
       configurer = VCAP::CloudController::Security::SecurityContextConfigurer.new(token_decoder)
 
       logger = access_log(config)
 
       Rack::Builder.new do
-        use CloudFoundry::Middleware::RequestMetrics, request_metrics
         use CloudFoundry::Middleware::Cors, config.get(:allowed_cors_domains)
         use CloudFoundry::Middleware::VcapRequestContextSetter
         use CloudFoundry::Middleware::BelowMinCliWarning if config.get(:warn_if_below_min_cli_version)
         use CloudFoundry::Middleware::NewRelicCustomAttributes if config.get(:newrelic_enabled)
         use CloudFoundry::Middleware::SecurityContextSetter, configurer
         use CloudFoundry::Middleware::Zipkin
-        use CloudFoundry::Middleware::RequestLogs, request_logs
+
+        if config.get(:concurrency_rate_limiter, :enabled)
+          use CloudFoundry::Middleware::ConcurrencyRateLimiter, {
+            logger: Steno.logger('cc.concurrency_rate_limiter'),
+            blocking_limit: config.get(:concurrency_rate_limiter, :blocking_limit),
+            logging_limit: config.get(:concurrency_rate_limiter, :logging_limit),
+            redis_connection_pool_size: config.get(:redis_connection_pool_size),
+            redis_counter_ttl_seconds: config.get(:redis_counter_ttl_seconds)
+          }
+        end
+
         if config.get(:rate_limiter, :enabled)
           use CloudFoundry::Middleware::RateLimiter, {
             logger: Steno.logger('cc.rate_limiter'),
@@ -45,7 +57,9 @@ def build(config, request_metrics, request_logs)
           use CloudFoundry::Middleware::ServiceBrokerRateLimiter, {
             logger: Steno.logger('cc.service_broker_rate_limiter'),
             max_concurrent_requests: config.get(:max_concurrent_service_broker_requests),
-            broker_timeout_seconds: config.get(:broker_client_timeout_seconds)
+            broker_timeout_seconds: config.get(:broker_client_timeout_seconds),
+            redis_connection_pool_size: config.get(:redis_connection_pool_size),
+            redis_counter_ttl_seconds: config.get(:redis_counter_ttl_seconds)
           }
         end
         if config.get(:rate_limiter_v2_api, :enabled)
@@ -59,6 +73,10 @@ def build(config, request_metrics, request_logs)
           }
         end
 
+        use CloudFoundry::Middleware::RequestMetrics, request_metrics
+        use CloudFoundry::Middleware::UserContextSetter, configurer
+        use CloudFoundry::Middleware::RequestLogs, request_logs
+
         use CloudFoundry::Middleware::CefLogs, Logger.new(config.get(:security_event_logging, :file)), config.get(:local_route) if config.get(:security_event_logging, :enabled)
         use Rack::CommonLogger, logger if logger
 
@@ -76,6 +94,7 @@ def build(config, request_metrics, request_logs)
         end
       end
     end
+    # rubocop:enable Metrics/MethodLength, Metrics/BlockLength
 
     private
 

lib/cloud_controller/security/security_context_configurer.rb

@@ -6,14 +6,29 @@ def initialize(token_decoder)
       end
 
       def configure(header_token)
+        configure_token_only(header_token)
+        configure_user
+      rescue VCAP::CloudController::UaaTokenDecoder::BadToken
+        VCAP::CloudController::SecurityContext.set(nil, :invalid_token, header_token)
+      end
+
+      def configure_token_only(header_token)
         VCAP::CloudController::SecurityContext.clear
         decoded_token = decode_token(header_token)
+        VCAP::CloudController::SecurityContext.set(nil, decoded_token, header_token)
+      rescue VCAP::CloudController::UaaTokenDecoder::BadToken
+        VCAP::CloudController::SecurityContext.set(nil, :invalid_token, header_token)
+      end
 
+      def configure_user
+        return unless VCAP::CloudController::SecurityContext.valid_token?
+
+        decoded_token = VCAP::CloudController::SecurityContext.token
         user = user_from_token(decoded_token)
         set_is_oauth_client(user, decoded_token)
-        VCAP::CloudController::SecurityContext.set(user, decoded_token, header_token)
+        VCAP::CloudController::SecurityContext.set(user, decoded_token, VCAP::CloudController::SecurityContext.auth_token)
       rescue VCAP::CloudController::UaaTokenDecoder::BadToken
-        VCAP::CloudController::SecurityContext.set(nil, :invalid_token, header_token)
+        VCAP::CloudController::SecurityContext.set(nil, :invalid_token, VCAP::CloudController::SecurityContext.auth_token)
       end
 
       private

middleware/base_rate_limiter.rb

@@ -1,4 +1,6 @@
-require 'mixins/client_ip'
+require 'mixins/user_id'
+require 'mixins/too_many_requests'
+require 'mixins/basic_auth'
 require 'mixins/user_reset_interval'
 require 'redis'
 
@@ -89,7 +91,9 @@ def to_hash
     end
 
     class BaseRateLimiter
-      include CloudFoundry::Middleware::ClientIp
+      include CloudFoundry::Middleware::UserId
+      include CloudFoundry::Middleware::TooManyRequests
+      include CloudFoundry::Middleware::BasicAuth
 
       def initialize(app, logger, expiring_request_counter, reset_interval, header_suffix=nil)
         @app = app
@@ -111,17 +115,13 @@ def call(env)
           rate_limit_headers.reset = (Time.now.to_i + expires_in).to_s
           rate_limit_headers.remaining = estimate_remaining(env, count)
 
-          return too_many_requests!(expires_in, env, rate_limit_headers) if exceeded_rate_limit(count, env)
+          return too_many_requests!(env, rate_limit_error_name(env), retry_after: expires_in, extra_headers: rate_limit_headers.to_hash) if exceeded_rate_limit(count, env)
         end
 
         status, headers, body = @app.call(env)
         [status, headers.merge(rate_limit_headers.to_hash), body]
       end
 
-      def get_user_id(env)
-        user_token?(env) ? env['cf.user_guid'] : client_ip(ActionDispatch::Request.new(env))
-      end
-
       private
 
       def apply_rate_limiting?(_env)
@@ -132,7 +132,7 @@ def global_request_limit(_env)
         raise 'method should be implemented in concrete class'
       end
 
-      def rate_limit_error(_env)
+      def rate_limit_error_name(_env)
         raise 'method should be implemented in concrete class'
       end
 
@@ -153,27 +153,9 @@ def estimate_remaining(env, new_count)
         [0, estimate].max.to_i.to_s
       end
 
-      def user_token?(env)
-        !!env['cf.user_guid']
-      end
-
-      def basic_auth?(env)
-        auth = Rack::Auth::Basic::Request.new(env)
-        auth.provided? && auth.basic?
-      end
-
       def admin?
         VCAP::CloudController::SecurityContext.admin? || VCAP::CloudController::SecurityContext.admin_read_only?
       end
-
-      def too_many_requests!(expires_in, env, rate_limit_headers)
-        headers = {}
-        headers['Retry-After'] = expires_in.to_s
-        headers['Content-Type'] = 'text/plain; charset=utf-8'
-        message = rate_limit_error(env).to_json
-        headers['Content-Length'] = message.length.to_s
-        [429, rate_limit_headers.to_hash.merge(headers), [message]]
-      end
     end
   end
 end

middleware/concurrency_rate_limiter.rb

@@ -0,0 +1,58 @@
+require 'mixins/user_id'
+require 'mixins/internal_or_root_api'
+require 'mixins/too_many_requests'
+require 'mixins/basic_auth'
+require 'concurrent_request_counter'
+
+module CloudFoundry
+  module Middleware
+    class ConcurrencyRateLimiter
+      include CloudFoundry::Middleware::UserId
+      include CloudFoundry::Middleware::InternalOrRootApi
+      include CloudFoundry::Middleware::TooManyRequests
+      include CloudFoundry::Middleware::BasicAuth
+
+      def initialize(app, opts)
+        @app = app
+        @logger = opts[:logger]
+        @counter = ConcurrentRequestCounter.instance(
+          'concurrent-rate-limit',
+          blocking_limit: opts[:blocking_limit],
+          logging_limit: opts[:logging_limit],
+          redis_connection_pool_size: opts[:redis_connection_pool_size],
+          redis_counter_ttl_seconds: opts[:redis_counter_ttl_seconds]
+        )
+      end
+
+      def call(env)
+        user_guid = nil
+        decrement_after_call = false
+
+        if apply_rate_limiting?(env)
+          user_guid = get_user_id(env)
+          decrement_after_call = @counter.try_increment?(user_guid, @logger)
+          return too_many_requests!(env, rate_limit_error_name(env), retry_after: suggested_retry_after) unless decrement_after_call
+        end
+
+        @app.call(env)
+      ensure
+        @counter.decrement(user_guid, @logger) if decrement_after_call
+      end
+
+      private
+
+      def suggested_retry_after
+        rand(1..5).to_i
+      end
+
+      def apply_rate_limiting?(env)
+        request = ActionDispatch::Request.new(env)
+        !basic_auth?(env) && !internal_api?(request) && !root_api?(request)
+      end
+
+      def rate_limit_error_name(env)
+        user_token?(env) ? 'ConcurrentRequestLimitExceeded' : 'IPBasedConcurrentRequestLimitExceeded'
+      end
+    end
+  end
+end