Enable per-api rate limits (#356)

api_loader.go

@@ -292,6 +292,7 @@ func processSpec(spec *APISpec, apisByListen map[string]int,
 		mwAppendEnabled(&chainArray, &RateCheckMW{BaseMiddleware: baseMid})
 		mwAppendEnabled(&chainArray, &IPWhiteListMiddleware{BaseMiddleware: baseMid})
 		mwAppendEnabled(&chainArray, &OrganizationMonitor{BaseMiddleware: baseMid})
+		mwAppendEnabled(&chainArray, &RateLimitForAPI{BaseMiddleware: baseMid})
 		mwAppendEnabled(&chainArray, &MiddlewareContextVars{BaseMiddleware: baseMid})
 		mwAppendEnabled(&chainArray, &VersionCheck{BaseMiddleware: baseMid})
 		mwAppendEnabled(&chainArray, &RequestSizeLimitMiddleware{baseMid})
@@ -444,6 +445,7 @@ func processSpec(spec *APISpec, apisByListen map[string]int,
 
 		mwAppendEnabled(&chainArray, &KeyExpired{baseMid})
 		mwAppendEnabled(&chainArray, &AccessRightsCheck{baseMid})
+		mwAppendEnabled(&chainArray, &RateLimitForAPI{BaseMiddleware: baseMid})
 		mwAppendEnabled(&chainArray, &RateLimitAndQuotaCheck{baseMid})
 		mwAppendEnabled(&chainArray, &GranularAccessMiddleware{baseMid})
 		mwAppendEnabled(&chainArray, &TransformMiddleware{baseMid})

apidef/api_definitions.go

@@ -352,6 +352,12 @@ type APIDefinition struct {
 	Tags              []string               `bson:"tags" json:"tags"`
 	EnableContextVars bool                   `bson:"enable_context_vars" json:"enable_context_vars"`
 	ConfigData        map[string]interface{} `bson:"config_data" json:"config_data"`
+	GlobalRateLimit   GlobalRateLimit        `bson:"global_rate_limit" json:"global_rate_limit"`
+}
+
+type GlobalRateLimit struct {
+	Rate float64 `bson:"rate" json:"rate"`
+	Per  float64 `bson:"per" json:"per"`
 }
 
 type BundleManifest struct {

mw_api_rate_limit.go

@@ -0,0 +1,74 @@
+package main
+
+import (
+	"errors"
+	"fmt"
+	"github.com/Sirupsen/logrus"
+	"net/http"
+)
+
+// RateLimitAndQuotaCheck will check the incomming request and key whether it is within it's quota and
+// within it's rate limit, it makes use of the SessionLimiter object to do this
+type RateLimitForAPI struct {
+	BaseMiddleware
+	keyName string
+	apiSess *SessionState
+}
+
+func (k *RateLimitForAPI) Name() string {
+	return "RateLimitForAPI"
+}
+
+func (k *RateLimitForAPI) EnabledForSpec() bool {
+	if k.Spec.DisableRateLimit || k.Spec.GlobalRateLimit.Rate == 0 {
+		return false
+	}
+
+	// We'll init here
+	k.keyName = fmt.Sprintf("apilimiter-%s%s", k.Spec.OrgID, k.Spec.APIID)
+	k.apiSess = &SessionState{
+		Rate:        k.Spec.GlobalRateLimit.Rate,
+		Per:         k.Spec.GlobalRateLimit.Per,
+		LastUpdated: "na",
+	}
+
+	return true
+}
+
+func (k *RateLimitForAPI) handleRateLimitFailure(r *http.Request, token string) (error, int) {
+	log.WithFields(logrus.Fields{
+		"path":   r.URL.Path,
+		"origin": requestIP(r),
+		"key":    token,
+	}).Info("API rate limit exceeded.")
+
+	// Fire a rate limit exceeded event
+	k.FireEvent(EventRateLimitExceeded, EventRateLimitExceededMeta{
+		EventMetaDefault: EventMetaDefault{Message: "API Rate Limit Exceeded", OriginatingRequest: EncodeRequestToEvent(r)},
+		Path:             r.URL.Path,
+		Origin:           requestIP(r),
+		Key:              token,
+	})
+
+	// Report in health check
+	reportHealthValue(k.Spec, Throttle, "-1")
+
+	return errors.New("API Rate limit exceeded"), 429
+}
+
+// ProcessRequest will run any checks on the request on the way through the system, return an error to have the chain fail
+func (k *RateLimitForAPI) ProcessRequest(w http.ResponseWriter, r *http.Request, _ interface{}) (error, int) {
+	storeRef := k.Spec.SessionManager.Store()
+	reason := sessionLimiter.ForwardMessage(k.apiSess,
+		k.keyName,
+		storeRef,
+		true,
+		false)
+
+	if reason == sessionFailRateLimit {
+		return k.handleRateLimitFailure(r, k.keyName)
+	}
+
+	// Request is valid, carry on
+	return nil, 200
+}

mw_api_rate_limit_test.go

@@ -0,0 +1,154 @@
+package main
+
+import (
+	"github.com/justinas/alice"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+	"time"
+)
+
+func createRLSession() *SessionState {
+	session := new(SessionState)
+	// essentially non-throttled
+	session.Rate = 100.0
+	session.Allowance = session.Rate
+	session.LastCheck = time.Now().Unix()
+	session.Per = 1.0
+	session.QuotaRenewalRate = 300 // 5 minutes
+	session.QuotaRenews = time.Now().Unix()
+	session.QuotaRemaining = 10
+	session.QuotaMax = 10
+	session.AccessRights = map[string]AccessDefinition{"31": {APIName: "Tyk Auth Key Test", APIID: "31", Versions: []string{"default"}}}
+	return session
+}
+
+func getRLOpenChain(spec *APISpec) http.Handler {
+	remote, _ := url.Parse(spec.Proxy.TargetURL)
+	proxy := TykNewSingleHostReverseProxy(remote, spec)
+	proxyHandler := ProxyHandler(proxy, spec)
+	baseMid := BaseMiddleware{spec, proxy}
+	chain := alice.New(mwList(
+		&IPWhiteListMiddleware{baseMid},
+		&RateLimitForAPI{BaseMiddleware: baseMid},
+		&VersionCheck{BaseMiddleware: baseMid},
+	)...).Then(proxyHandler)
+	return chain
+}
+
+func getGlobalRLAuthKeyChain(spec *APISpec) http.Handler {
+	remote, _ := url.Parse(spec.Proxy.TargetURL)
+	proxy := TykNewSingleHostReverseProxy(remote, spec)
+	proxyHandler := ProxyHandler(proxy, spec)
+	baseMid := BaseMiddleware{spec, proxy}
+	chain := alice.New(mwList(
+		&IPWhiteListMiddleware{baseMid},
+		&AuthKey{baseMid},
+		&VersionCheck{BaseMiddleware: baseMid},
+		&KeyExpired{baseMid},
+		&AccessRightsCheck{baseMid},
+		&RateLimitForAPI{BaseMiddleware: baseMid},
+		&RateLimitAndQuotaCheck{baseMid},
+	)...).Then(proxyHandler)
+	return chain
+}
+
+func TestRLOpen(t *testing.T) {
+	spec := createSpecTest(t, openRLDefSmall)
+
+	req := testReq(t, "GET", "/rl_test/", nil)
+
+	DRLManager.CurrentTokenValue = 1
+	DRLManager.RequestTokenValue = 1
+
+	chain := getRLOpenChain(spec)
+	for a := 0; a <= 10; a++ {
+		recorder := httptest.NewRecorder()
+		chain.ServeHTTP(recorder, req)
+		if a < 3 {
+			if recorder.Code != 200 {
+				t.Fatalf("Rate limit kicked in too early, after only %v requests", a)
+			}
+		}
+
+		if a > 7 {
+			if recorder.Code != 429 {
+				t.Fatalf("Rate limit did not activate, code was: %v", recorder.Code)
+			}
+		}
+	}
+}
+
+func TestRLClosed(t *testing.T) {
+	spec := createSpecTest(t, closedRLDefSmall)
+
+	req := testReq(t, "GET", "/rl_closed_test/", nil)
+
+	session := createRLSession()
+	customToken := "54321111"
+	// AuthKey sessions are stored by {token}
+	spec.SessionManager.UpdateSession(customToken, session, 60)
+	req.Header.Set("authorization", "Bearer "+customToken)
+
+	DRLManager.CurrentTokenValue = 1
+	DRLManager.RequestTokenValue = 1
+
+	chain := getGlobalRLAuthKeyChain(spec)
+	for a := 0; a <= 10; a++ {
+		recorder := httptest.NewRecorder()
+		chain.ServeHTTP(recorder, req)
+		if a < 3 {
+			if recorder.Code != 200 {
+				t.Fatalf("Rate limit kicked in too early, after only %v requests", a)
+			}
+		}
+
+		if a > 7 {
+			if recorder.Code != 429 {
+				t.Fatalf("Rate limit did not activate, code was: %v", recorder.Code)
+			}
+		}
+	}
+}
+
+const openRLDefSmall = `{
+	"api_id": "31",
+	"org_id": "default",
+	"auth": {"auth_header_name": "authorization"},
+	"use_keyless": true,
+	"version_data": {
+		"not_versioned": true,
+		"versions": {
+			"v1": {"name": "v1"}
+		}
+	},
+	"proxy": {
+		"listen_path": "/rl_test/",
+		"target_url": "` + testHttpAny + `"
+	},
+	"global_rate_limit": {
+		"rate": 3,
+		"per": 1
+	}
+}`
+
+const closedRLDefSmall = `{
+	"api_id": "31",
+	"org_id": "default",
+	"auth": {"auth_header_name": "authorization"},
+	"version_data": {
+		"not_versioned": true,
+		"versions": {
+			"v1": {"name": "v1"}
+		}
+	},
+	"proxy": {
+		"listen_path": "/rl_closed_test/",
+		"target_url": "` + testHttpAny + `"
+	},
+	"global_rate_limit": {
+		"rate": 3,
+		"per": 1
+	}
+}`

session_manager.go

@@ -117,7 +117,6 @@ func (l *SessionLimiter) ForwardMessage(currentSession *SessionState, key string
 				return sessionFailRateLimit
 			}
 
-			//log.Info("Add is: ", DRLManager.CurrentTokenValue)
 			_, errF := userBucket.Add(uint(DRLManager.CurrentTokenValue))
 
 			if errF != nil {