SignalR rate throttling

[replaysMike]

Is your feature request related to a problem? Please describe.

I'm looking to rate-throttle SignalR websocket requests to prevent abuse of the API. There doesn't appear to be middleware or event hooks in SignalR in order to process each message.

Describe the solution you'd like

I'd like to see websocket requests running through ASP.Net middleware so messages can be intercepted. Or at the very least a SignalR specific event hook for pre and post processing of websocket messages.

Describe alternatives you've considered

The only alternative currently that I've been able to use is at the Hub level. For each endpoint, I can run the request through a rate throttler but this is not ideal. I can't even implement this as an attribute level filter (that I'm aware of) it's manual code on every endpoint.

[replaysMike]

The alternative I've had to use roll my own throttling, but it has to be done manually in each endpoint of the Hub. What I'd like to see is a way to intercept all messages (WebSocket especially) before a response is sent to the client. In my case I used the rate limiting provided by AspNetCoreRateLimit package, with a custom provider. I can't use the package as is because it needs to be able to modify the response, but the response is already created when it hits the endpoints so modifying the response code or headers is not allowed.

public enum ThrottleOption {
  /// <summary>
  /// Default behavior, returns a BadRequestResponse when throttling occurs
  /// </summary>
  None,
  /// <summary>
  /// Throw an exception when throttling occurs
  /// </summary>
  ThrowOnError
}

/// <summary>
/// SignalR rate throttling provider
/// </summary>
public class SignalRRateThrottleProvider {
  private IpRateLimitOptions _options;
  private IIpAddressParser _ipParser;
  private IpRateLimitProcessor _processor;
  private ILogger<SignalRRateThrottleProvider> _logger;

  /// <summary>
  /// Create a SignalR rate throttling provider
  /// </summary>
  /// <param name="logger"></param>
  /// <param name="rateLimit"></param>
  /// <param name="options"></param>
  /// <param name="ipParser"></param>
  public SignalRRateThrottleProvider (ILogger<SignalRRateThrottleProvider> logger, IpRateLimitProcessor rateLimit, IOptions<IpRateLimitOptions> options, IIpAddressParser ipParser) {
    _logger = logger;
    _options = options != null ? options.Value : null;
    _ipParser = ipParser;
    _processor = rateLimit;
  }

  /// <summary>
  /// Perform rate throttling on a SignalR request
  /// </summary>
  /// <param name="callerContext"></param>
  /// <returns></returns>
  public async Task<IActionResult> ThrottleAsync (HubCallerContext callerContext) {
    return await ThrottleAsync (callerContext, ThrottleOption.None);
  }

  /// <summary>
  /// Perform rate throttling on a SignalR request
  /// </summary>
  /// <param name="callerContext"></param>
  /// <param name="options">The throttle options to use</param>
  /// <returns></returns>
  public async Task<IActionResult> ThrottleAsync (HubCallerContext callerContext, ThrottleOption options) {
    if (callerContext != null) {
      var context = callerContext.GetHttpContext ();
      var identity = SetIdentity (context);
      if (_processor.IsWhitelisted (identity)) {
        // allow through
        return null;
      }

      var rules = _processor.GetMatchingRules (identity);
      return await ProcessRulesAsync (rules, identity, callerContext.GetHttpContext (), options);
    }

    // no context available, allow through
    return null;
  }

  private async Task<IActionResult> ProcessRulesAsync (List<RateLimitRule> rules, ClientRequestIdentity identity, HttpContext httpContext, ThrottleOption options) {
    foreach (var rule in rules) {
      if (rule.Limit > 0) {
        // increment counter
        var counter = _processor.ProcessRequest (identity, rule);

        // check if key expired
        if (counter.Timestamp + rule.PeriodTimespan.Value < DateTime.UtcNow) {
          continue;
        }

        // check if limit is reached
        if (counter.TotalRequests > rule.Limit) {
          //compute retry after value
          var retryAfter = _processor.RetryAfterFrom (counter.Timestamp, rule);

          // log blocked request
          LogBlockedRequest (httpContext, identity, counter, rule);

          // break execution
          return await ReturnQuotaExceededResponseAsync (httpContext, rule, retryAfter, options);
        }
      }
      // if limit is zero or less, block the request.
      else {
        // process request count
        var counter = _processor.ProcessRequest (identity, rule);

        // log blocked request
        LogBlockedRequest (httpContext, identity, counter, rule);

        // break execution (Int32 max used to represent infinity)
        return await ReturnQuotaExceededResponseAsync (httpContext, rule, Int32.MaxValue.ToString (System.Globalization.CultureInfo.InvariantCulture), options);
      }
    }

    return null;
  }

  private ClientRequestIdentity SetIdentity (HttpContext httpContext) {
    var clientId = "anon";
    if (httpContext.Request.Headers.Keys.Contains (_options.ClientIdHeader, StringComparer.CurrentCultureIgnoreCase)) {
      clientId = httpContext.Request.Headers[_options.ClientIdHeader].First ();
    }

    var clientIp = string.Empty;
    try {
      var ip = _ipParser.GetClientIp (httpContext);
      if (ip == null) {
        throw new Exception ("IpRateLimitMiddleware can't parse caller IP");
      }

      clientIp = ip.ToString ();
    } catch (Exception ex) {
      throw new Exception ("IpRateLimitMiddleware can't parse caller IP", ex);
    }

    return new ClientRequestIdentity {
      ClientIp = clientIp,
        Path = httpContext.Request.Path.ToString ().ToLowerInvariant (),
        HttpVerb = httpContext.Request.Method.ToLowerInvariant (),
        ClientId = clientId
    };
  }

  private async Task<IActionResult> ReturnQuotaExceededResponseAsync (HttpContext httpContext, RateLimitRule rule, string retryAfter, ThrottleOption options) {
    var message = string.IsNullOrEmpty (_options.QuotaExceededMessage) ? $"API calls quota exceeded! Maximum admitted {rule.Limit} per {rule.Period}." : _options.QuotaExceededMessage;
    var ex = new ThrottledException (message);
    if (options.HasFlag (ThrottleOption.ThrowOnError))
      throw ex;

    return new ApiBadRequestResponse (ex);
  }

  private void LogBlockedRequest (HttpContext httpContext, ClientRequestIdentity identity, RateLimitCounter counter, RateLimitRule rule) {
    _logger.LogInformation ($"Request {identity.HttpVerb}:{identity.Path} from IP {identity.ClientIp} has been blocked, quota {rule.Limit}/{rule.Period} exceeded by {counter.TotalRequests}. Blocked by rule {rule.Endpoint}, TraceIdentifier {httpContext.TraceIdentifier}.");
  }
}

and used as follows in the SignalR Hub - MyHub.cs

public async Task<IActionResult> TestEndpointAsync(MyRequest request)
{
    var response = await _rateThrottleProvider.ThrottleAsync(Context);
    if (response != null)
        return response;
    return await MyService.ProcessRequest(Context, request);
}

Clearly it would be more ideal to be able to either decorate the hub with a request filter, or even better provide a delegate or middleware capable of processing each request message.

[davidfowl]

So are you trying to rate limit the websocket connection itself or individual messsges? I ask because you say “before a response is sent to the client” which response is that?

[replaysMike]

Yes I'm trying to rate limit the websocket connection. In other words, the rate/frequency of incoming websocket messages over the connection.

The response is modified when throttling occurs, which means I need to intercept the request and prevent it from going further. If it exceeds max rates it would need to modify the response with an appropriate error. It's not ideal to do this on each endpoint, but there doesn't seem to be any hooks in place to accomplish this.

[aspnet-hello]

We periodically close 'discussion' issues that have not been updated in a long period of time.

We apologize if this causes any inconvenience. We ask that if you are still encountering an issue, please log a new issue with updated information and we will investigate.