Add concurrent request rate limiter and optimize middleware stack

config/cloud_controller.yml

@@ -275,6 +275,14 @@ rate_limiter_v2_api:
   global_admin_limit: 20000
   reset_interval_in_minutes: 60
 
+concurrency_rate_limiter:
+  enabled: false
+  blocking_limit: 10
+  logging_limit: 10
+  redis_connection_pool_size: 40
+  redis_counter_ttl_seconds: 60
+
+
 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_schemas/api_schema.rb

@@ -395,6 +395,14 @@ class ApiSchema < VCAP::Config
             reset_interval_in_minutes: Integer
           },
 
+          optional(:concurrency_rate_limiter) => {
+            enabled: bool,
+            optional(:blocking_limit) => Integer,
+            optional(:logging_limit) => Integer,
+            optional(:redis_connection_pool_size) => Integer,
+            optional(:redis_counter_ttl_seconds) => 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(:concurrency_rate_limiter, :redis_connection_pool_size),
+            redis_counter_ttl_seconds: config.get(:concurrency_rate_limiter, :redis_counter_ttl_seconds)
+          }
+        end
+
         if config.get(:rate_limiter, :enabled)
           use CloudFoundry::Middleware::RateLimiter, {
             logger: Steno.logger('cc.rate_limiter'),
@@ -59,6 +71,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 +92,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,31 @@ def initialize(token_decoder)
       end
 
       def configure(header_token)
+        configure_token_only(header_token)
+        return unless VCAP::CloudController::SecurityContext.valid_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_token_only(decoded_token, header_token)
+      rescue VCAP::CloudController::UaaTokenDecoder::BadToken
+        VCAP::CloudController::SecurityContext.set(nil, :invalid_token, header_token)
+      end
+
+      def configure_user
+        decoded_token = VCAP::CloudController::SecurityContext.token
+        return unless decoded_token && decoded_token != :invalid_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

lib/cloud_controller/security_context.rb

@@ -12,6 +12,12 @@ def self.set(user, token=nil, auth_token=nil)
       Thread.current[:vcap_auth_token] = auth_token
     end
 
+    def self.set_token_only(token, auth_token=nil)
+      Thread.current[:vcap_user] = nil
+      Thread.current[:vcap_token] = token
+      Thread.current[:vcap_auth_token] = auth_token
+    end
+
     def self.current_user
       Thread.current[:vcap_user]
     end

middleware/concurrency_rate_limiter.rb

@@ -0,0 +1,228 @@
+require 'mixins/client_ip'
+
+module CloudFoundry
+  module Middleware
+    class StoreError < StandardError; end
+
+    class ConcurrentRedisStore
+      def initialize(redis, counter_ttl_seconds: nil)
+        @redis = redis
+        @counter_ttl_seconds = counter_ttl_seconds
+      end
+
+      def self.new_socket(socket, connection_pool_size: nil, counter_ttl_seconds: nil)
+        connection_pool_size ||= VCAP::CloudController::Config.config.get(:puma, :max_threads) || 1
+        redis = ConnectionPool::Wrapper.new(size: connection_pool_size) do
+          Redis.new(timeout: 1, path: socket)
+        end
+        new(redis, counter_ttl_seconds: counter_ttl_seconds)
+      end
+
+      def increment(key, logger)
+        count = @redis.incr(key).to_i
+        @redis.expire(key, @counter_ttl_seconds) if @counter_ttl_seconds
+        count
+      rescue Redis::BaseError => e
+        logger.error("Redis error: #{e.class} - #{e.message}")
+        raise StoreError.new("increment failed: #{e.message}")
+      end
+
+      def decrement(key, logger)
+        count = @redis.decr(key).to_i
+        @redis.incr(key) if count < 0
+        [count, 0].max
+      rescue Redis::BaseError => e
+        logger.error("Redis error: #{e.class} - #{e.message}")
+        raise StoreError.new("decrement failed: #{e.message}")
+      end
+    end
+
+    class ConcurrentInMemoryStore
+      def initialize
+        @mutex = Mutex.new
+        @data = {}
+      end
+
+      def increment(key, _logger)
+        @mutex.synchronize do
+          @data[key] = (@data[key] || 0) + 1
+        end
+      end
+
+      def decrement(key, _logger)
+        @mutex.synchronize do
+          return 0 unless @data.key?(key)
+
+          @data[key] -= 1
+          @data.delete(key) if @data[key] <= 0
+          @data[key] || 0
+        end
+      end
+    end
+
+    class ConcurrencyLimiter
+      @instance_mutex = Mutex.new
+
+      def self.instance(logger, blocking_limit: nil, logging_limit: nil, redis_connection_pool_size: nil, redis_counter_ttl_seconds: nil)
+        return @instance if @instance
+
+        @instance_mutex.synchronize do
+          @instance ||= new(logger,
+                            blocking_limit: blocking_limit,
+                            logging_limit: logging_limit,
+                            redis_connection_pool_size: redis_connection_pool_size,
+                            redis_counter_ttl_seconds: redis_counter_ttl_seconds)
+        end
+        @instance
+      end
+
+      def initialize(logger, blocking_limit: nil, logging_limit: nil, redis_connection_pool_size: nil, redis_counter_ttl_seconds: nil)
+        @blocking_limit = blocking_limit
+        @logging_limit = logging_limit
+        @redis_connection_pool_size = redis_connection_pool_size
+        @redis_counter_ttl_seconds = redis_counter_ttl_seconds
+        @logger = logger
+      end
+
+      def try_increment?(user_guid)
+        return true unless @blocking_limit&.>=(0) || @logging_limit&.>=(0)
+
+        key = "#{key_prefix}:#{user_guid}"
+        count = store.increment(key, @logger)
+
+        if @logging_limit&.>=(0) && count > @logging_limit
+          @logger.info("Concurrency limit warning for user '#{user_guid}', count=#{count} exceeded logging_limit=#{@logging_limit}")
+        end
+
+        if @blocking_limit&.>=(0) && count > @blocking_limit
+          store.decrement(key, @logger)
+          @logger.info("Concurrent rate limit exceeded for user '#{user_guid}', limit=#{@blocking_limit} remaining=0")
+          return false
+        end
+
+        true
+      rescue StoreError
+        # fail open
+        true
+      end
+
+      def decrement(user_guid)
+        return unless @blocking_limit&.>=(0) || @logging_limit&.>=(0)
+
+        key = "#{key_prefix}:#{user_guid}"
+        store.decrement(key, @logger)
+      rescue StoreError
+        # fail open
+      end
+
+      def suggested_retry_after
+        rand(1..5).to_i
+      end
+
+      def error_name
+        'ConcurrentRequestLimitExceeded'
+      end
+
+      def error_name_ip_based
+        'IPBasedConcurrentRequestLimitExceeded'
+      end
+
+      private
+
+      def key_prefix
+        'concurrent-rate-limit'
+      end
+
+      def store
+        return @store if defined?(@store)
+
+        redis_socket = VCAP::CloudController::Config.config.get(:redis, :socket)
+        @store = if redis_socket.nil?
+                   ConcurrentInMemoryStore.new
+                 else
+                   ConcurrentRedisStore.new_socket(redis_socket, connection_pool_size: @redis_connection_pool_size, counter_ttl_seconds: @redis_counter_ttl_seconds)
+                 end
+      end
+    end
+
+    class ConcurrencyRateLimiter
+      include CloudFoundry::Middleware::ClientIp
+
+      def initialize(app, opts)
+        @app = app
+        @logger = opts[:logger]
+        @concurrency_limiter = ConcurrencyLimiter.instance(
+          opts[:logger],
+          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
+        incremented = false
+
+        if apply_rate_limiting?(env)
+          user_guid = get_user_id(env)
+          incremented = @concurrency_limiter.try_increment?(user_guid)
+          return too_many_requests!(env) unless incremented
+        end
+
+        status, headers, body = @app.call(env)
+        [status, headers, body]
+      ensure
+        @concurrency_limiter.decrement(user_guid) if incremented
+      end
+
+      private
+
+      def get_user_id(env)
+        user_token?(env) ? env['cf.user_guid'] : client_ip(ActionDispatch::Request.new(env))
+      end
+
+      def user_token?(env)
+        !!env['cf.user_guid']
+      end
+
+      def too_many_requests!(env)
+        headers = {}
+        headers['Retry-After'] = @concurrency_limiter.suggested_retry_after.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, headers, [message]]
+      end
+
+      def apply_rate_limiting?(env)
+        request = ActionDispatch::Request.new(env)
+        !basic_auth?(env) && !internal_api?(request) && !root_api?(request)
+      end
+
+      def root_api?(request)
+        request.fullpath.match(%r{\A(?:/v2/info|/v3|/|/healthz)\z})
+      end
+
+      def internal_api?(request)
+        request.fullpath.match(%r{\A/internal})
+      end
+
+      def basic_auth?(env)
+        auth = Rack::Auth::Basic::Request.new(env)
+        auth.provided? && auth.basic?
+      end
+
+      def rate_limit_error(env)
+        error_name = user_token?(env) ? @concurrency_limiter.error_name : @concurrency_limiter.error_name_ip_based
+        api_error = CloudController::Errors::ApiError.new_from_details(error_name)
+        version   = env['PATH_INFO'][0..2]
+        if version == '/v2'
+          ErrorPresenter.new(api_error, Rails.env.test?, V2ErrorHasher.new(api_error)).to_hash
+        elsif version == '/v3'
+          ErrorPresenter.new(api_error, Rails.env.test?, V3ErrorHasher.new(api_error)).to_hash
+        end
+      end
+    end
+  end
+end

middleware/security_context_setter.rb

@@ -29,7 +29,7 @@ def call(env)
           end
         end
 
-        security_context_configurer.configure(header_token)
+        security_context_configurer.configure_token_only(header_token)
 
         if VCAP::CloudController::SecurityContext.valid_token?
           env['cf.user_guid'] = id_from_token